diff --git a/test/check_ledger.tsv b/test/check_ledger.tsv
index a9fdef60b..10900f8bb 100644
--- a/test/check_ledger.tsv
+++ b/test/check_ledger.tsv
@@ -2560,6 +2560,37 @@ iceberg_fdw_projection	iceberg_fdw_projection	one column : FDW projection matche
 iceberg_fdw_projection	iceberg_fdw_projection	select one, filter another : FDW projection matches iceberg_scan	15;16;17;18;19	2026-10-21	the Iceberg FDW's projection mask ignores columns only the scan clauses reference
 iceberg_fdw_projection	iceberg_fdw_projection	three columns : FDW projection matches iceberg_scan	15;16;17;18;19	never	-
 iceberg_fdw_projection	iceberg_fdw_projection	whole row : FDW projection matches iceberg_scan	15;16;17;18;19	never	-
+iceberg_malformed	iceberg_malformed	a NULL data-file path is refused in iceberg_data_files (XX001, not a crash)	15;16;17;18;19	2026-10-21	columnar_iceberg.c: a manifest entry with a NULL file_path is not refused
+iceberg_malformed	iceberg_malformed	a NULL data-file path is refused in iceberg_scan (XX001, not a crash)	15;16;17;18;19	2026-10-21	columnar_iceberg.c: a manifest entry with a NULL file_path is not refused
+iceberg_malformed	iceberg_malformed	a NULL manifest_path is refused, not a crash (XX001)	15;16;17;18;19	2026-10-21	columnar_iceberg.c: ice_open_path does not refuse a NULL recorded path
+iceberg_malformed	iceberg_malformed	a dangling current-schema-id is refused (XX001), not silently stale-read	15;16;17;18;19	2026-10-21	columnar_iceberg.c: a dangling current-schema-id resolves to the legacy top-level schema
+iceberg_malformed	iceberg_malformed	a manifest-list that is a FIFO is refused, not a hang (XX001)	15;16;17;18;19	2026-10-21	columnar_objstore.c: PgColumnarOpenLocalRegularFile drops its S_ISREG guard, so a FIFO opens O_NONBLOCK and reads as an empty file
+iceberg_malformed	iceberg_malformed	a null manifest-list min_sequence_number decodes as NULL, not 0 (#686)	15;16;17;18;19	2026-10-21	columnar_avro.c: read_manifest_list decodes a null min_sequence_number as 0
+iceberg_malformed	iceberg_malformed	a null manifest-list sequence_number decodes as NULL, not 0	15;16;17;18;19	2026-10-21	columnar_avro.c: read_manifest_list decodes a null sequence_number as 0
+iceberg_malformed	iceberg_malformed	an array-valued schema fields element is refused, not crashed (XX001)	15;16;17;18;19	never	-
+iceberg_malformed	iceberg_malformed	backend still up after the FIFO refusals	15;16;17;18;19	never	-
+iceberg_malformed	iceberg_malformed	backend still up after the dangling-schema refusal	15;16;17;18;19	never	-
+iceberg_malformed	iceberg_malformed	backend still up after the lister refusal	15;16;17;18;19	never	-
+iceberg_malformed	iceberg_malformed	backend still up after the malformed-schema refusal	15;16;17;18;19	never	-
+iceberg_malformed	iceberg_malformed	backend still up after the manifest-list FIFO refusal	15;16;17;18;19	never	-
+iceberg_malformed	iceberg_malformed	backend still up after the null-manifest_path refusal	15;16;17;18;19	2026-10-21	columnar_iceberg.c: ice_open_path does not refuse a NULL recorded path
+iceberg_malformed	iceberg_malformed	backend still up after the null-sequence refusal	15;16;17;18;19	never	-
+iceberg_malformed	iceberg_malformed	backend still up after the scan refusal	15;16;17;18;19	2026-10-21	columnar_iceberg.c: a manifest entry with a NULL file_path is not refused
+iceberg_malformed	iceberg_malformed	control: a concrete manifest seq scans all 5 rows (seq-3 delete is older than seq-5 data)	15;16;17;18;19	never	-
+iceberg_malformed	iceberg_malformed	control: a legacy-only-schema table still reads (5 rows)	15;16;17;18;19	never	-
+iceberg_malformed	iceberg_malformed	control: a present file_path lists one data file	15;16;17;18;19	never	-
+iceberg_malformed	iceberg_malformed	control: a present manifest-list sequence_number decodes to its value	15;16;17;18;19	never	-
+iceberg_malformed	iceberg_malformed	control: a present min_sequence_number decodes to its value (#686)	15;16;17;18;19	never	-
+iceberg_malformed	iceberg_malformed	control: a resolvable current-schema-id reads the correct amount column	15;16;17;18;19	never	-
+iceberg_malformed	iceberg_malformed	control: a well-formed field object still decodes (0 entries)	15;16;17;18;19	never	-
+iceberg_malformed	iceberg_malformed	control: the legacy table binds amount correctly (10..50)	15;16;17;18;19	never	-
+iceberg_malformed	iceberg_malformed	iceberg_current_snapshot on a FIFO is refused, not a hang (XX001)	15;16;17;18;19	2026-10-21	columnar_objstore.c: PgColumnarOpenLocalRegularFile drops its S_ISREG guard, so a FIFO opens O_NONBLOCK and reads as an empty file
+iceberg_malformed	iceberg_malformed	iceberg_data_files on a FIFO is refused, not a hang (XX001)	15;16;17;18;19	2026-10-21	columnar_objstore.c: PgColumnarOpenLocalRegularFile drops its S_ISREG guard, so a FIFO opens O_NONBLOCK and reads as an empty file
+iceberg_malformed	iceberg_malformed	iceberg_scan on a FIFO is refused, not a hang (XX001)	15;16;17;18;19	2026-10-21	columnar_objstore.c: PgColumnarOpenLocalRegularFile drops its S_ISREG guard, so a FIFO opens O_NONBLOCK and reads as an empty file
+iceberg_malformed	iceberg_malformed	iceberg_scan refuses a null manifest seq an ADDED entry must inherit (XX001, not a silent wrong row)	15;16;17;18;19	2026-10-21	columnar_iceberg.c: an added entry inherits a null manifest sequence number as 0
+iceberg_malformed	iceberg_malformed	read_avro_manifest on a FIFO is refused, not a hang (XX001)	15;16;17;18;19	2026-10-21	columnar_objstore.c: PgColumnarOpenLocalRegularFile drops its S_ISREG guard, so a FIFO opens O_NONBLOCK and reads as an empty file
+iceberg_malformed	iceberg_malformed	read_manifest_list on a FIFO is refused, not a hang (XX001)	15;16;17;18;19	2026-10-21	columnar_objstore.c: PgColumnarOpenLocalRegularFile drops its S_ISREG guard, so a FIFO opens O_NONBLOCK and reads as an empty file
+iceberg_malformed	iceberg_malformed	read_parquet on a FIFO is refused, not a hang (XX001)	15;16;17;18;19	2026-10-21	columnar_objstore.c: PgColumnarOpenLocalRegularFile drops its S_ISREG guard, so a FIFO opens O_NONBLOCK and reads as an empty file
 iceberg_name_mapping_memory	iceberg_name_mapping_memory	and the mapped values are still correct	15;16;17;18;19	2026-10-21	the name-mapping parse binds every name but region, so region reads NULL on every row;the name-mapping parse ignores every entry's names, so nothing binds
 iceberg_name_mapping_memory	iceberg_name_mapping_memory	core's own parse is the larger part of the scan's peak	15;16;17;18;19	2026-10-21	the name-mapping parse ignores every entry's names, so nothing binds
 iceberg_name_mapping_memory	iceberg_name_mapping_memory	the large arm produced three readings	15;16;17;18;19	2026-10-21	the name-mapping parse ignores every entry's names, so nothing binds
diff --git a/test/check_ledger_budget.txt b/test/check_ledger_budget.txt
index 07074f0c9..66efc310d 100644
--- a/test/check_ledger_budget.txt
+++ b/test/check_ledger_budget.txt
@@ -15,7 +15,7 @@
 #
 #   Adding a check to a suite that is already covered does not move it, which is
 #   what makes it safe to bound.
-suites_not_covered 105
+suites_not_covered 104
 #
 # checks_never_observed_red -- A CENSUS. NOT a ceiling, and it must not become
 #   one.
@@ -5374,4 +5374,22 @@ suites_not_covered 105
 #     never         3850 + 3 = 3853
 #     dated         2239 + 4 = 2243
 #     3853 + 2243 == 6096
-checks_never_observed_red 3853
+# #1341, iceberg_malformed seeded, with its twin (OffgridwithJD): 3853 -> 3868,
+# and suites_not_covered 105 -> 104. The suite had no rows. It enters with 31
+# checks from five-major runs (all green on each of 15-19, shell and twin alike), after
+# two fixes measured on PG17 at origin/main. Every refusal arm read only XX001, which
+# later parse steps also raise: with the regular-file guard removed, a FIFO opened
+# O_NONBLOCK reads as an empty file and four of the seven FIFO arms still passed. Each
+# arm now also names its refusal's message. And every "backend still up" check was a
+# bare SELECT 1, which answers again after crash recovery: with a NULL path segfaulting
+# the backend and the probe 3s later, both survival checks passed. They now read the
+# server log's crash count since their own arm, and wait out a recovery. 16 are dated
+# by seven mutations, each reddening the same names in both twins; 15 arrive never.
+# The array-valued "fields" arm stays never: on a build without asserts the schema
+# walker still refuses through av_schema_err, so dropping the object check passes all 31.
+#
+#     ledger rows   6096 + 31 = 6127
+#     never         3853 + 15 = 3868
+#     dated         2243 + 16 = 2259
+#     3868 + 2259 == 6127
+checks_never_observed_red 3868
diff --git a/test/iceberg_malformed.sh b/test/iceberg_malformed.sh
index 872dd228a..588f62063 100755
--- a/test/iceberg_malformed.sh
+++ b/test/iceberg_malformed.sh
@@ -5,8 +5,10 @@
 # whoever authored the table, and is called by an honest pg_read_server_files
 # role that named only the trusted metadata pointer. A crafted manifest must be
 # REFUSED cleanly, never crash the backend. Each arm has a control so a vacuous
-# pass is visible, and a "backend still up" probe after every refusal proves the
-# refusal was an error, not a segfault. Fixtures are hand-crafted
+# pass is visible, each refusal names its own message (the SQLSTATE alone is
+# shared with later parse errors), and a "backend still up" probe after every
+# refusal reads the server log's crash count, so a segfault followed by crash
+# recovery does not pass as a refusal. Fixtures are hand-crafted
 # (test/fixtures/iceberg/warehouse_malformed) because no writer emits these
 # shapes; see gen_malformed_fixture.py.
 #
@@ -16,31 +18,69 @@ set -uo pipefail
 . "$(dirname "${BASH_SOURCE[0]}")/lib.sh"
 pgc_setup "${1:-/usr/local/pg17/bin/pg_config}"
 
-sqlstate_of() {
+# refusal SQL REGEX LABEL -> "<SQLSTATE> (LABEL)" when SQL raised an error whose message
+# matches REGEX, "<SQLSTATE>: <message>" for any other error, or "no error". Every
+# arm here refuses with XX001, and so do later parse steps: a FIFO read as an empty
+# file failed a parse with the same XX001, so the SQLSTATE alone passed with the
+# guard removed. The arm now names the refusal it wants.
+refusal_of() {	# stdin: psql output -> the refusal reading
+	local re="$1" label="$2" line state msg
+	local out; out="$(cat)"
+	line="$(sed -n 's/^ERROR:  \([0-9A-Z]\{5\}\): /\1|/p' <<<"$out" | head -1)"
+	if [ -z "$line" ]; then
+		grep -qE 'server closed the connection|connection to server' <<<"$out" \
+			&& echo "no refusal: the connection was lost" || echo "no error"
+		return
+	fi
+	state="${line%%|*}"; msg="${line#*|}"
+	if grep -qE -- "$re" <<<"$msg"; then echo "$state ($label)"; else echo "$state: $msg"; fi
+}
+refusal() {
 	env PATH="$PGC_BINDIR:$PATH" psql -h 127.0.0.1 -p "$PGC_PORT" -U postgres \
-		-d "$PGC_DB" -qtA 2>&1 <<SQLEOF | sed -n 's/^ERROR:  \([0-9A-Z]\{5\}\).*/\1/p' | head -1
-\\set VERBOSITY sqlstate
+		-d "$PGC_DB" -qtA 2>&1 <<SQLEOF | refusal_of "$2" "$3"
+\\set VERBOSITY verbose
 $1;
 SQLEOF
 }
 
 # Run one statement under a HARD wall-clock cap. A cancel-resistant open (a FIFO)
 # cannot be ended by statement_timeout, so the external `timeout` is the only
-# reliable detector. Prints the 5-char SQLSTATE the backend raised, or the
-# literal HANG when psql was KILLed by the cap (exit 124/137). No SQL-level
+# reliable detector. Prints the refusal reading as refusal() does, or the literal
+# HANG when psql was KILLed by the cap (exit 124/137). No SQL-level
 # statement_timeout is set, so a merely-slow legit path is never a false HANG.
 FIFO_CAP=5
-sqlstate_or_hang() {
+FIFO_RE='is not a regular file'
+refusal_or_hang() {
 	local out rc
 	out="$(timeout -s KILL "$FIFO_CAP" env PATH="$PGC_BINDIR:$PATH" psql \
 		-h 127.0.0.1 -p "$PGC_PORT" -U postgres -d "$PGC_DB" -qtA 2>&1 <<SQLEOF
-\\set VERBOSITY sqlstate
+\\set VERBOSITY verbose
 $1;
 SQLEOF
 )"
 	rc=$?
 	if [ "$rc" -eq 124 ] || [ "$rc" -eq 137 ]; then echo HANG; return; fi
-	printf '%s\n' "$out" | sed -n 's/^ERROR:  \([0-9A-Z]\{5\}\).*/\1/p' | head -1
+	printf '%s\n' "$out" | refusal_of "$FIFO_RE" "not a regular file"
+}
+
+# Backends killed by a signal, from the server log: a number, or nothing when the
+# log can't be read. A fresh connection answers again after crash recovery, so
+# "SELECT 1" alone passed after a backend segfaulted on a NULL path, once the probe
+# came 3s later. Every survival check also requires that this count has not moved
+# since just before its own arm ran; when it has, the check waits out the crash
+# recovery, so the arms after it are not collateral.
+crashes() {
+	local _n
+	_n="$(pgc_pg "grep -c 'terminated by signal' '$PGC_LOGFILE'" 2>/dev/null)"
+	case "$_n" in '' | *[!0-9]*) ;; *) echo "$_n" ;; esac
+}
+survived() {	# survived BEFORE -> "1" when the server answers and no backend died since BEFORE
+	local _up _now _i
+	_up="$(q 'SELECT 1')"; _now="$(crashes)"
+	if ! [[ "$1" =~ ^[0-9]+$ && "$_now" =~ ^[0-9]+$ ]]; then echo "no (crash count unreadable)"; return; fi
+	if [ "$_up" = 1 ] && [ "$_now" = "$1" ]; then echo 1; return; fi
+	for _i in $(seq 1 60); do [ "$(q 'SELECT 1' 2>/dev/null)" = 1 ] && break; sleep 0.5; done
+	echo "no ([$_up], $(( _now - $1 )) backend crashes)"
 }
 
 # Release a backend that is blocked in open(2) on a FIFO (main only): opening the
@@ -70,10 +110,12 @@ chmod -R u+rwX "$DEST"
 #     backend (an object-container Assert / OOB heap read). It must now be a
 #     clean malformed-schema error, and the control must still decode.
 # =========================================================================
-check "an array-valued schema \"fields\" element is refused, not crashed (XX001)" \
-	"$(sqlstate_of "SELECT * FROM pgcolumnar.read_avro_manifest('$DEST/arrayfield/evil.avro')")" \
-	"XX001"
-check "backend still up after the malformed-schema refusal" "$(q 'SELECT 1')" "1"
+B="$(crashes)"
+check "an array-valued schema fields element is refused, not crashed (XX001)" \
+	"$(refusal "SELECT * FROM pgcolumnar.read_avro_manifest('$DEST/arrayfield/evil.avro')" \
+		'unsupported or malformed Avro schema in manifest' "malformed schema")" \
+	"XX001 (malformed schema)"
+check "backend still up after the malformed-schema refusal" "$(survived "$B")" "1"
 check "control: a well-formed field object still decodes (0 entries)" \
 	"$(q "SELECT count(*) FROM pgcolumnar.read_avro_manifest('$DEST/arrayfield/good.avro')")" \
 	"0"
@@ -84,12 +126,16 @@ check "control: a well-formed field object still decodes (0 entries)" \
 #     be a clean DATA_CORRUPTED refusal; the control (file_path present) reads.
 # =========================================================================
 NP="$DEST/nullpath/db/t/metadata/t.metadata.json"
+B="$(crashes)"
 check "a NULL data-file path is refused in iceberg_data_files (XX001, not a crash)" \
-	"$(sqlstate_of "SELECT * FROM pgcolumnar.iceberg_data_files('$NP')")" "XX001"
-check "backend still up after the lister refusal" "$(q 'SELECT 1')" "1"
+	"$(refusal "SELECT * FROM pgcolumnar.iceberg_data_files('$NP')" \
+		'has no data file path' "no data file path")" "XX001 (no data file path)"
+check "backend still up after the lister refusal" "$(survived "$B")" "1"
+B="$(crashes)"
 check "a NULL data-file path is refused in iceberg_scan (XX001, not a crash)" \
-	"$(sqlstate_of "SELECT * FROM pgcolumnar.iceberg_scan('$NP') AS t(id bigint)")" "XX001"
-check "backend still up after the scan refusal" "$(q 'SELECT 1')" "1"
+	"$(refusal "SELECT * FROM pgcolumnar.iceberg_scan('$NP') AS t(id bigint)" \
+		'has no data file path' "no data file path")" "XX001 (no data file path)"
+check "backend still up after the scan refusal" "$(survived "$B")" "1"
 check "control: a present file_path lists one data file" \
 	"$(q "SELECT count(*) FROM pgcolumnar.iceberg_data_files('$DEST/nullpath_ctl/db/t/metadata/t.metadata.json')")" \
 	"1"
@@ -101,9 +147,11 @@ check "control: a present file_path lists one data file" \
 #     author-controlled, so this is reachable. Refuse it as DATA_CORRUPTED.
 # =========================================================================
 NMP="$DEST/nullmanifestpath/db/t/metadata/t.metadata.json"
+B="$(crashes)"
 check "a NULL manifest_path is refused, not a crash (XX001)" \
-	"$(sqlstate_of "SELECT * FROM pgcolumnar.iceberg_data_files('$NMP')")" "XX001"
-check "backend still up after the null-manifest_path refusal" "$(q 'SELECT 1')" "1"
+	"$(refusal "SELECT * FROM pgcolumnar.iceberg_data_files('$NMP')" \
+		'a manifest path in .* is null' "null manifest path")" "XX001 (null manifest path)"
+check "backend still up after the null-manifest_path refusal" "$(survived "$B")" "1"
 
 # =========================================================================
 # #1  a non-regular file (FIFO) named on any Iceberg-reachable open must be
@@ -113,31 +161,32 @@ check "backend still up after the null-manifest_path refusal" "$(q 'SELECT 1')"
 FIFODIR="$PGC_WORKDIR/fifo"; mkdir -p "$FIFODIR"
 FIFO="$FIFODIR/x"; mkfifo "$FIFO"
 
+B="$(crashes)"
 # Site A -- ice_slurp_text, FIFO as the metadata argument (three entry points)
 check "iceberg_current_snapshot on a FIFO is refused, not a hang (XX001)" \
-	"$(sqlstate_or_hang "SELECT * FROM pgcolumnar.iceberg_current_snapshot('$FIFO')")" \
-	"XX001"; fifo_release "$FIFO"
+	"$(refusal_or_hang "SELECT * FROM pgcolumnar.iceberg_current_snapshot('$FIFO')")" \
+	"XX001 (not a regular file)"; fifo_release "$FIFO"
 check "iceberg_data_files on a FIFO is refused, not a hang (XX001)" \
-	"$(sqlstate_or_hang "SELECT * FROM pgcolumnar.iceberg_data_files('$FIFO')")" \
-	"XX001"; fifo_release "$FIFO"
+	"$(refusal_or_hang "SELECT * FROM pgcolumnar.iceberg_data_files('$FIFO')")" \
+	"XX001 (not a regular file)"; fifo_release "$FIFO"
 check "iceberg_scan on a FIFO is refused, not a hang (XX001)" \
-	"$(sqlstate_or_hang "SELECT * FROM pgcolumnar.iceberg_scan('$FIFO') AS t(id bigint)")" \
-	"XX001"; fifo_release "$FIFO"
+	"$(refusal_or_hang "SELECT * FROM pgcolumnar.iceberg_scan('$FIFO') AS t(id bigint)")" \
+	"XX001 (not a regular file)"; fifo_release "$FIFO"
 
 # Site C -- av_slurp_file, FIFO as the argument (both entry points)
 check "read_avro_manifest on a FIFO is refused, not a hang (XX001)" \
-	"$(sqlstate_or_hang "SELECT * FROM pgcolumnar.read_avro_manifest('$FIFO')")" \
-	"XX001"; fifo_release "$FIFO"
+	"$(refusal_or_hang "SELECT * FROM pgcolumnar.read_avro_manifest('$FIFO')")" \
+	"XX001 (not a regular file)"; fifo_release "$FIFO"
 check "read_manifest_list on a FIFO is refused, not a hang (XX001)" \
-	"$(sqlstate_or_hang "SELECT * FROM pgcolumnar.read_manifest_list('$FIFO')")" \
-	"XX001"; fifo_release "$FIFO"
+	"$(refusal_or_hang "SELECT * FROM pgcolumnar.read_manifest_list('$FIFO')")" \
+	"XX001 (not a regular file)"; fifo_release "$FIFO"
 
 # Site D -- pq_source_open_cfg via the single-file parquet API (sibling)
 check "read_parquet on a FIFO is refused, not a hang (XX001)" \
-	"$(sqlstate_or_hang "SELECT * FROM pgcolumnar.read_parquet('$FIFO') AS t(id bigint)")" \
-	"XX001"; fifo_release "$FIFO"
+	"$(refusal_or_hang "SELECT * FROM pgcolumnar.read_parquet('$FIFO') AS t(id bigint)")" \
+	"XX001 (not a regular file)"; fifo_release "$FIFO"
 
-check "backend still up after the FIFO refusals" "$(q 'SELECT 1')" "1"
+check "backend still up after the FIFO refusals" "$(survived "$B")" "1"
 
 # Site B -- ice_slurp_bin, FIFO named as the manifest-list INSIDE metadata.json.
 # realpath() on the FIFO succeeds and passes ice_open_path's containment, so only
@@ -153,10 +202,11 @@ cat > "$MDJSON" <<JSON
    "schema-id":0,"manifest-list":"file://$MLROOT/db/t/metadata/ml.avro",
    "summary":{"operation":"append"}}]}
 JSON
+B="$(crashes)"
 check "a manifest-list that is a FIFO is refused, not a hang (XX001)" \
-	"$(sqlstate_or_hang "SELECT * FROM pgcolumnar.iceberg_data_files('$MDJSON')")" \
-	"XX001"; fifo_release "$MLFIFO"
-check "backend still up after the manifest-list FIFO refusal" "$(q 'SELECT 1')" "1"
+	"$(refusal_or_hang "SELECT * FROM pgcolumnar.iceberg_data_files('$MDJSON')")" \
+	"XX001 (not a regular file)"; fifo_release "$MLFIFO"
+check "backend still up after the manifest-list FIFO refusal" "$(survived "$B")" "1"
 
 # =========================================================================
 # #4  a manifest_file.sequence_number that is the union's NULL branch decoded
@@ -182,10 +232,12 @@ check "control: a present min_sequence_number decodes to its value (#686)" \
 
 NS="$DEST/nullseq/db/t/metadata/t.metadata.json"
 NSC="$DEST/nullseq_ctl/db/t/metadata/t.metadata.json"
+B="$(crashes)"
 check "iceberg_scan refuses a null manifest seq an ADDED entry must inherit (XX001, not a silent wrong row)" \
-	"$(sqlstate_of "SELECT count(*) FROM pgcolumnar.iceberg_scan('$NS') AS t(id bigint, region text, amount int)")" \
-	"XX001"
-check "backend still up after the null-sequence refusal" "$(q 'SELECT 1')" "1"
+	"$(refusal "SELECT count(*) FROM pgcolumnar.iceberg_scan('$NS') AS t(id bigint, region text, amount int)" \
+		'has a null sequence number that an added entry' "null sequence to inherit")" \
+	"XX001 (null sequence to inherit)"
+check "backend still up after the null-sequence refusal" "$(survived "$B")" "1"
 check "control: a concrete manifest seq scans all 5 rows (seq-3 delete is older than seq-5 data)" \
 	"$(q "SELECT count(*) FROM pgcolumnar.iceberg_scan('$NSC') AS t(id bigint, region text, amount int)")" \
 	"5"
@@ -197,10 +249,12 @@ check "control: a concrete manifest seq scans all 5 rows (seq-3 delete is older
 #     schema and misprojects every row). Asymmetric with ice_current_snapshot.
 # =========================================================================
 DS="$DEST/danglingschema/db/t/metadata/t.metadata.json"
+B="$(crashes)"
 check "a dangling current-schema-id is refused (XX001), not silently stale-read" \
-	"$(sqlstate_of "SELECT amount FROM pgcolumnar.iceberg_scan('$DS') AS t(amount int)")" \
-	"XX001"
-check "backend still up after the dangling-schema refusal" "$(q 'SELECT 1')" "1"
+	"$(refusal "SELECT amount FROM pgcolumnar.iceberg_scan('$DS') AS t(amount int)" \
+		'is not present in the schemas array' "dangling schema id")" \
+	"XX001 (dangling schema id)"
+check "backend still up after the dangling-schema refusal" "$(survived "$B")" "1"
 # control: the SAME data with a RESOLVABLE current-schema-id reads the correct
 # column (amount = field id 3 = 10..50). Proves only the dangling id flips it.
 DSC="$DEST/danglingschema_ctl/db/t/metadata/t.metadata.json"
diff --git a/test/pytest/TESTS.md b/test/pytest/TESTS.md
index 74f2e000b..95399115e 100644
--- a/test/pytest/TESTS.md
+++ b/test/pytest/TESTS.md
@@ -310,6 +310,7 @@ behaviour, the source of that number is named.
 - [262. test_parallel_flush_optin.py: parallel_flush stays opt-in, and the dispatch metric cannot separate the shapes](#262-test_parallel_flush_optinpy-parallel_flush-stays-opt-in-and-the-dispatch-metric-cannot-separate-the-shapes)
 - [263. test_parquet_export_stats.py: the Parquet files we export carry correct per-row-group statistics](#263-test_parquet_export_statspy-the-parquet-files-we-export-carry-correct-per-row-group-statistics)
 - [264. test_catalog_filter_reads.py: stats() and sort_status() read a table's storage id once](#264-test_catalog_filter_readspy-stats-and-sort_status-read-a-tables-storage-id-once)
+- [265. test_iceberg_malformed.py: the Iceberg reader refuses crafted metadata cleanly, never crashing or hanging](#265-test_iceberg_malformedpy-the-iceberg-reader-refuses-crafted-metadata-cleanly-never-crashing-or-hanging)
 
 ## 1. How to read a test in here
 
@@ -10270,3 +10271,47 @@ The pytest twin of `test/catalog_filter_reads.sh` (#1637), every check by the sh
 | `test_stats_reports_only_the_tables_groups` | control: stats() reports t's 3 row groups and no other table's |
 | `test_sort_status_reads_the_storage_id_once` | sort_status() reads the storage id once, not once per catalog row |
 | `test_sort_status_counts_only_the_tables_groups` | control: sort_status() counts t's 3 row groups and no other table's |
+
+## 265. test_iceberg_malformed.py: the Iceberg reader refuses crafted metadata cleanly, never crashing or hanging
+
+The twin of `test/iceberg_malformed.sh` (#388 phase 4a, #644). The hand-crafted fixtures in `test/fixtures/iceberg/warehouse_malformed` are refused with the refusal each arm names, not merely its SQLSTATE: an array-valued Avro `fields` element, a NULL data-file path in the lister and the scan, a NULL manifest_path, a FIFO at every Iceberg-reachable open (a HANG after 5s), a null manifest sequence number an added entry must inherit, and a dangling current-schema-id. Null manifest-list sequence numbers decode as NULL, and every refusal has a control that still reads.
+
+### Every test
+
+One module fixture copies the fixtures into `server_dir`, builds the FIFOs there, and records every reading from a private database. A FIFO read runs in a thread with a 5s deadline. Each survival reading compares the server log's crash count with the one taken just before its arm, and waits out a crash recovery. The first two tests mirror the shell's two skip names.
+
+| test | what it holds |
+| --- | --- |
+| `test_the_malformed_iceberg_fixtures_are_present` | premise: the committed malformed fixtures are not empty |
+| `test_python_reads_the_malformed_metadata` | premise: the dangling-schema metadata parses as JSON |
+| `test_an_array_valued_fields_element_is_refused` | an array-valued schema fields element is refused, not crashed (XX001) |
+| `test_the_backend_survives_the_malformed_schema` | backend still up after the malformed-schema refusal |
+| `test_a_well_formed_field_object_still_decodes` | control: a well-formed field object still decodes (0 entries) |
+| `test_the_lister_refuses_a_null_data_file_path` | a NULL data-file path is refused in iceberg_data_files (XX001, not a crash) |
+| `test_the_backend_survives_the_lister_refusal` | backend still up after the lister refusal |
+| `test_the_scan_refuses_a_null_data_file_path` | a NULL data-file path is refused in iceberg_scan (XX001, not a crash) |
+| `test_the_backend_survives_the_scan_refusal` | backend still up after the scan refusal |
+| `test_a_present_file_path_lists_one_data_file` | control: a present file_path lists one data file |
+| `test_a_null_manifest_path_is_refused` | a NULL manifest_path is refused, not a crash (XX001) |
+| `test_the_backend_survives_the_null_manifest_path` | backend still up after the null-manifest_path refusal |
+| `test_iceberg_current_snapshot_refuses_a_fifo` | iceberg_current_snapshot on a FIFO is refused, not a hang (XX001) |
+| `test_iceberg_data_files_refuses_a_fifo` | iceberg_data_files on a FIFO is refused, not a hang (XX001) |
+| `test_iceberg_scan_refuses_a_fifo` | iceberg_scan on a FIFO is refused, not a hang (XX001) |
+| `test_read_avro_manifest_refuses_a_fifo_as_iceberg_input` | read_avro_manifest on a FIFO is refused, not a hang (XX001) |
+| `test_read_manifest_list_refuses_a_fifo` | read_manifest_list on a FIFO is refused, not a hang (XX001) |
+| `test_read_parquet_refuses_a_fifo_beside_iceberg` | read_parquet on a FIFO is refused, not a hang (XX001) |
+| `test_the_backend_survives_the_fifo_refusals` | backend still up after the FIFO refusals |
+| `test_a_manifest_list_that_is_a_fifo_is_refused` | a manifest-list that is a FIFO is refused, not a hang (XX001) |
+| `test_the_backend_survives_the_manifest_list_fifo` | backend still up after the manifest-list FIFO refusal |
+| `test_a_null_manifest_list_sequence_number_reads_null` | a null manifest-list sequence_number decodes as NULL, not 0 |
+| `test_a_present_manifest_list_sequence_number_reads_its_value` | control: a present manifest-list sequence_number decodes to its value |
+| `test_a_null_min_sequence_number_reads_null` | a null manifest-list min_sequence_number decodes as NULL, not 0 (#686) |
+| `test_a_present_min_sequence_number_reads_its_value` | control: a present min_sequence_number decodes to its value (#686) |
+| `test_the_scan_refuses_a_null_sequence_an_added_entry_inherits` | iceberg_scan refuses a null manifest seq an ADDED entry must inherit (XX001, not a silent wrong row) |
+| `test_the_backend_survives_the_null_sequence_refusal` | backend still up after the null-sequence refusal |
+| `test_a_concrete_manifest_sequence_scans_every_row` | control: a concrete manifest seq scans all 5 rows (seq-3 delete is older than seq-5 data) |
+| `test_a_dangling_current_schema_id_is_refused` | a dangling current-schema-id is refused (XX001), not silently stale-read |
+| `test_the_backend_survives_the_dangling_schema_refusal` | backend still up after the dangling-schema refusal |
+| `test_a_resolvable_current_schema_id_reads_amount` | control: a resolvable current-schema-id reads the correct amount column |
+| `test_a_legacy_only_schema_table_still_reads` | control: a legacy-only-schema table still reads (5 rows) |
+| `test_the_legacy_table_binds_amount` | control: the legacy table binds amount correctly (10..50) |
diff --git a/test/pytest/expected_tests.txt b/test/pytest/expected_tests.txt
index c4c6e9a81..cffb8bff7 100644
--- a/test/pytest/expected_tests.txt
+++ b/test/pytest/expected_tests.txt
@@ -1025,6 +1025,8 @@ guard_tests 474
 # 5 tests, one per check, by collection over the cluster files.
 # cluster_tests 2508 -> 2670 for #1341's test_parquet_export_stats.py, the twin of parquet_export_stats.sh:
 # 162 tests, one per check, by collection over the cluster files.
+# cluster_tests 2677 -> 2710 for #1341's test_iceberg_malformed.py, the twin of iceberg_malformed.sh:
+# 31 checks in 33 tests, by collection over the cluster files.
 # cluster_tests 2670 -> 2677 for #1637's test_catalog_filter_reads.py, the twin of catalog_filter_reads.sh:
 # 7 tests, one per check, by collection over the cluster files.
-cluster_tests 2677
+cluster_tests 2710
diff --git a/test/pytest/test_compare_to_bash.py b/test/pytest/test_compare_to_bash.py
index f6c63c000..d89e8525b 100644
--- a/test/pytest/test_compare_to_bash.py
+++ b/test/pytest/test_compare_to_bash.py
@@ -124,6 +124,7 @@
             "iceberg_name_mapping_memory",
             "iceberg_catalog",
             "iceberg_scan",
+            "iceberg_malformed",
             "concurrency",
             "isolation",
             "index_only",
diff --git a/test/pytest/test_iceberg_malformed.py b/test/pytest/test_iceberg_malformed.py
new file mode 100644
index 000000000..5b966150b
--- /dev/null
+++ b/test/pytest/test_iceberg_malformed.py
@@ -0,0 +1,446 @@
+"""The Iceberg reader refuses crafted metadata, manifests and manifest lists cleanly,
+never crashing or hanging the backend (#388 phase 4a, #644). The pytest twin of
+`test/iceberg_malformed.sh`, every check by the shell's name (#1341).
+
+The hand-crafted fixtures under test/fixtures/iceberg/warehouse_malformed are copied
+into `server_dir`, so the recorded root differs from the actual one and every read
+goes through the same path rebasing the honest read path uses. Each refusal is read
+as "<SQLSTATE> (<label>)" only when the error's primary message is the refusal under
+test: every arm refuses with XX001, and so do later parse steps, so a FIFO read as an
+empty file passed on its SQLSTATE alone with the regular-file guard removed. A FIFO
+read that is still running after 5s is a HANG; its write end is then opened, so the
+backend is released.
+
+A BACKEND THAT DIED DOES NOT PASS AS ONE THAT SURVIVED. A fresh connection answers
+again after crash recovery, so every survival check also reads the server log's count
+of killed backends since the fixture began, and an unreadable log fails it.
+
+ON A PRIVATE DATABASE; each test asserts one reading (#1339).
+"""
+import json
+import os
+import pathlib
+import re
+import shutil
+import threading
+import time
+
+import psycopg
+import pytest
+
+from pgc_vacuity import query_error
+
+DB = "pgc_iceberg_malformed"
+WHM = pathlib.Path(__file__).resolve().parents[1] / "fixtures" / "iceberg" / "warehouse_malformed"
+NEEDED = ("nullpath/db/t/metadata/t.metadata.json", "nullseq_ml/ml.avro", "nullminseq_ml/ml.avro",
+          "nullmanifestpath/db/t/metadata/t.metadata.json", "danglingschema/db/t/metadata/t.metadata.json")
+# the refusal each arm wants, matched against the primary message as the shell does
+SCHEMA = (r"unsupported or malformed Avro schema in manifest", "malformed schema")
+NOPATH = (r"has no data file path", "no data file path")
+NULLMP = (r"a manifest path in .* is null", "null manifest path")
+FIFO_REFUSAL = (r"is not a regular file", "not a regular file")
+NULLSEQ = (r"has a null sequence number that an added entry", "null sequence to inherit")
+DANGLING = (r"is not present in the schemas array", "dangling schema id")
+
+
+def _message(exc):
+    return str(exc).splitlines()[0] if str(exc) else type(exc).__name__
+
+
+def _value(dsn, sql):
+    """-> the first column of the first row as the shell's psql prints it, or the
+    failure as a query_error()."""
+    try:
+        with psycopg.connect(dsn, autocommit=True) as conn:
+            row = conn.execute(sql).fetchone()
+    except psycopg.Error as exc:
+        return query_error(_message(exc))
+    if row is None or row[0] is None:
+        return query_error(f"no value from {sql}")
+    v = row[0]
+    return "t" if v is True else "f" if v is False else str(v)
+
+
+def _reading(exc, want):
+    """-> "<SQLSTATE> (<label>)" when the primary message is the wanted refusal,
+    "<SQLSTATE>: <message>" for any other error."""
+    if not exc.sqlstate:
+        return "no refusal: the connection was lost"
+    msg = exc.diag.message_primary or ""
+    regex, label = want
+    return f"{exc.sqlstate} ({label})" if re.search(regex, msg) else f"{exc.sqlstate}: {msg}"
+
+
+def _refusal(dsn, sql, want):
+    try:
+        with psycopg.connect(dsn, autocommit=True) as conn:
+            conn.execute(sql).fetchall()
+    except psycopg.Error as exc:
+        return _reading(exc, want)
+    return "no error"
+
+
+def _refusal_or_hang(dsn, sql, fifo):
+    """-> the refusal reading, or HANG when the statement is still running after 5s;
+    a hung backend is released by opening the FIFO's write end."""
+    result = {}
+
+    def run():
+        result["v"] = _refusal(dsn, sql, FIFO_REFUSAL)
+
+    t = threading.Thread(target=run, daemon=True)
+    t.start()
+    t.join(5)
+    if t.is_alive():
+        try:
+            os.close(os.open(fifo, os.O_WRONLY | os.O_NONBLOCK))
+        except OSError:
+            pass
+        t.join(10)
+        return "HANG"
+    return result.get("v", query_error("the statement left no outcome"))
+
+
+def _crashes(cluster):
+    try:
+        return (cluster.datadir / "server.log").read_text(errors="replace").count("terminated by signal")
+    except OSError as exc:
+        return query_error(f"crash count unreadable: {exc}")
+
+
+def _survived(dsn, cluster, before):
+    """-> "1" when the server answers and no backend died since `before`, read just
+    before the arm ran; after a crash, waits out the recovery so the arms after it
+    are not collateral."""
+    up, now = _value(dsn, "SELECT 1"), _crashes(cluster)
+    if not isinstance(before, int) or not isinstance(now, int):
+        return query_error(f"crash count unreadable: {before} / {now}")
+    if up == "1" and now == before:
+        return "1"
+    for _ in range(60):
+        if _value(dsn, "SELECT 1") == "1":
+            break
+        time.sleep(0.5)
+    return f"no ([{up if up == '1' else ''}], {now - before} backend crashes)"
+
+
+def _fixtures_present():
+    return all((WHM / f).is_file() for f in NEEDED)
+
+
+@pytest.fixture(scope="module")
+def imf(pgc_cluster, server_dir):
+    if not _fixtures_present():
+        yield None
+        return
+    dest = server_dir / "malformed"
+    shutil.copytree(WHM, dest)
+    for p in [dest, *dest.rglob("*")]:
+        p.chmod(0o777 if p.is_dir() else 0o666)
+    fifodir = server_dir / "fifo"
+    fifodir.mkdir()
+    fifodir.chmod(0o777)
+    fifo = fifodir / "x"
+    os.mkfifo(fifo)
+    os.chmod(fifo, 0o666)
+    mlroot = fifodir / "tbl"
+    (mlroot / "db" / "t" / "metadata").mkdir(parents=True)
+    for p in [mlroot, *mlroot.rglob("*")]:
+        p.chmod(0o777)
+    mlfifo = mlroot / "db" / "t" / "metadata" / "ml.avro"
+    os.mkfifo(mlfifo)
+    os.chmod(mlfifo, 0o666)
+    mdjson = mlroot / "db" / "t" / "metadata" / "t.metadata.json"
+    mdjson.write_text(json.dumps({
+        "format-version": 2, "location": f"file://{mlroot}/db/t",
+        "current-snapshot-id": 4242, "current-schema-id": 0,
+        "schemas": [{"schema-id": 0, "type": "struct", "fields": []}],
+        "snapshots": [{"snapshot-id": 4242, "timestamp-ms": 0, "sequence-number": 1, "schema-id": 0,
+                       "manifest-list": f"file://{mlroot}/db/t/metadata/ml.avro",
+                       "summary": {"operation": "append"}}]}))
+    mdjson.chmod(0o666)
+
+    with psycopg.connect(pgc_cluster.dsn(), autocommit=True) as a:
+        a.execute(f"DROP DATABASE IF EXISTS {DB} WITH (FORCE)")
+        a.execute(f"CREATE DATABASE {DB}")
+    try:
+        dsn = pgc_cluster.dsn(dbname=DB)
+        with psycopg.connect(dsn, autocommit=True) as c:
+            c.execute("CREATE EXTENSION IF NOT EXISTS pgcolumnar")
+        crashes = lambda: _crashes(pgc_cluster)  # noqa: E731
+        alive = lambda b: _survived(dsn, pgc_cluster, b)  # noqa: E731
+        md = lambda d: dest / d / "db" / "t" / "metadata" / "t.metadata.json"  # noqa: E731
+        scan3 = "AS t(id bigint, region text, amount int)"
+        r = {}
+        b = crashes()
+        r["array refused"] = _refusal(dsn, f"SELECT * FROM pgcolumnar.read_avro_manifest('{dest}/arrayfield/evil.avro')", SCHEMA)
+        r["array up"] = alive(b)
+        r["array control"] = _value(dsn, f"SELECT count(*) FROM pgcolumnar.read_avro_manifest('{dest}/arrayfield/good.avro')")
+        b = crashes()
+        r["np lister"] = _refusal(dsn, f"SELECT * FROM pgcolumnar.iceberg_data_files('{md('nullpath')}')", NOPATH)
+        r["np lister up"] = alive(b)
+        b = crashes()
+        r["np scan"] = _refusal(dsn, f"SELECT * FROM pgcolumnar.iceberg_scan('{md('nullpath')}') AS t(id bigint)", NOPATH)
+        r["np scan up"] = alive(b)
+        r["np control"] = _value(dsn, f"SELECT count(*) FROM pgcolumnar.iceberg_data_files('{md('nullpath_ctl')}')")
+        b = crashes()
+        r["nmp"] = _refusal(dsn, f"SELECT * FROM pgcolumnar.iceberg_data_files('{md('nullmanifestpath')}')", NULLMP)
+        r["nmp up"] = alive(b)
+        b = crashes()
+        for key, sql in (("fifo snapshot", f"SELECT * FROM pgcolumnar.iceberg_current_snapshot('{fifo}')"),
+                         ("fifo lister", f"SELECT * FROM pgcolumnar.iceberg_data_files('{fifo}')"),
+                         ("fifo scan", f"SELECT * FROM pgcolumnar.iceberg_scan('{fifo}') AS t(id bigint)"),
+                         ("fifo manifest", f"SELECT * FROM pgcolumnar.read_avro_manifest('{fifo}')"),
+                         ("fifo list", f"SELECT * FROM pgcolumnar.read_manifest_list('{fifo}')"),
+                         ("fifo parquet", f"SELECT * FROM pgcolumnar.read_parquet('{fifo}') AS t(id bigint)")):
+            r[key] = _refusal_or_hang(dsn, sql, fifo)
+        r["fifo up"] = alive(b)
+        b = crashes()
+        r["fifo ml"] = _refusal_or_hang(dsn, f"SELECT * FROM pgcolumnar.iceberg_data_files('{mdjson}')", mlfifo)
+        r["fifo ml up"] = alive(b)
+        r["seq null"] = _value(dsn, f"SELECT sequence_number IS NULL FROM pgcolumnar.read_manifest_list('{dest}/nullseq_ml/ml.avro')")
+        r["seq ctl"] = _value(dsn, f"SELECT sequence_number FROM pgcolumnar.read_manifest_list('{dest}/nullseq_ml/ml_ctl.avro')")
+        r["minseq null"] = _value(dsn, f"SELECT min_sequence_number IS NULL FROM pgcolumnar.read_manifest_list('{dest}/nullminseq_ml/ml.avro')")
+        r["minseq ctl"] = _value(dsn, f"SELECT min_sequence_number FROM pgcolumnar.read_manifest_list('{dest}/nullminseq_ml/ml_ctl.avro')")
+        b = crashes()
+        r["seq refused"] = _refusal(dsn, f"SELECT count(*) FROM pgcolumnar.iceberg_scan('{md('nullseq')}') {scan3}", NULLSEQ)
+        r["seq up"] = alive(b)
+        r["seq control"] = _value(dsn, f"SELECT count(*) FROM pgcolumnar.iceberg_scan('{md('nullseq_ctl')}') {scan3}")
+        b = crashes()
+        r["dangling"] = _refusal(dsn, f"SELECT amount FROM pgcolumnar.iceberg_scan('{md('danglingschema')}') AS t(amount int)", DANGLING)
+        r["dangling up"] = alive(b)
+        r["dangling control"] = _value(dsn, "SELECT string_agg(amount::text, ',' ORDER BY amount) "
+                                            f"FROM pgcolumnar.iceberg_scan('{md('danglingschema_ctl')}') AS t(amount int)")
+        r["legacy rows"] = _value(dsn, f"SELECT count(*) FROM pgcolumnar.iceberg_scan('{md('legacyschema')}') {scan3}")
+        r["legacy amount"] = _value(dsn, "SELECT string_agg(amount::text, ',' ORDER BY amount) "
+                                         f"FROM pgcolumnar.iceberg_scan('{md('legacyschema')}') {scan3}")
+        yield r
+    finally:
+        with psycopg.connect(pgc_cluster.dsn(), autocommit=True) as a:
+            a.execute(f"DROP DATABASE IF EXISTS {DB} WITH (FORCE)")
+
+
+def _declined(imf, expect):
+    if imf is not None:
+        return False
+    expect.cannot_run("ABSENT_FIXTURE", "the malformed iceberg fixtures are missing")
+    return True
+
+
+def test_the_malformed_iceberg_fixtures_are_present(expect):
+    """The shell skips the whole suite without the committed fixtures, under this name."""
+    if not _fixtures_present():
+        expect.cannot_run("MISSING_DEPENDENCY", "the committed malformed Iceberg fixtures are not in this tree",
+                          name="malformed iceberg fixtures are missing")
+        return
+    expect.at_least((WHM / NEEDED[0]).stat().st_size, 1,
+                    "premise: the committed malformed fixtures are not empty")
+
+
+def test_python_reads_the_malformed_metadata(expect):
+    """The shell's other skip name: it needs python3 with json."""
+    try:
+        import json as _json
+    except ImportError:
+        expect.cannot_run("MISSING_DEPENDENCY", "json is not importable", name="python3 is needed")
+        return
+    expect.at_least(len(_json.loads((WHM / NEEDED[4]).read_text())), 1,
+                    "premise: the dangling-schema metadata parses as JSON")
+
+
+def test_an_array_valued_fields_element_is_refused(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["array refused"], "XX001 (malformed schema)",
+                "an array-valued schema fields element is refused, not crashed (XX001)")
+
+
+def test_the_backend_survives_the_malformed_schema(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["array up"], "1", "backend still up after the malformed-schema refusal")
+
+
+def test_a_well_formed_field_object_still_decodes(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["array control"], "0", "control: a well-formed field object still decodes (0 entries)")
+
+
+def test_the_lister_refuses_a_null_data_file_path(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["np lister"], "XX001 (no data file path)",
+                "a NULL data-file path is refused in iceberg_data_files (XX001, not a crash)")
+
+
+def test_the_backend_survives_the_lister_refusal(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["np lister up"], "1", "backend still up after the lister refusal")
+
+
+def test_the_scan_refuses_a_null_data_file_path(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["np scan"], "XX001 (no data file path)",
+                "a NULL data-file path is refused in iceberg_scan (XX001, not a crash)")
+
+
+def test_the_backend_survives_the_scan_refusal(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["np scan up"], "1", "backend still up after the scan refusal")
+
+
+def test_a_present_file_path_lists_one_data_file(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["np control"], "1", "control: a present file_path lists one data file")
+
+
+def test_a_null_manifest_path_is_refused(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["nmp"], "XX001 (null manifest path)", "a NULL manifest_path is refused, not a crash (XX001)")
+
+
+def test_the_backend_survives_the_null_manifest_path(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["nmp up"], "1", "backend still up after the null-manifest_path refusal")
+
+
+def test_iceberg_current_snapshot_refuses_a_fifo(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["fifo snapshot"], "XX001 (not a regular file)",
+                "iceberg_current_snapshot on a FIFO is refused, not a hang (XX001)")
+
+
+def test_iceberg_data_files_refuses_a_fifo(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["fifo lister"], "XX001 (not a regular file)",
+                "iceberg_data_files on a FIFO is refused, not a hang (XX001)")
+
+
+def test_iceberg_scan_refuses_a_fifo(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["fifo scan"], "XX001 (not a regular file)", "iceberg_scan on a FIFO is refused, not a hang (XX001)")
+
+
+def test_read_avro_manifest_refuses_a_fifo_as_iceberg_input(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["fifo manifest"], "XX001 (not a regular file)",
+                "read_avro_manifest on a FIFO is refused, not a hang (XX001)")
+
+
+def test_read_manifest_list_refuses_a_fifo(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["fifo list"], "XX001 (not a regular file)",
+                "read_manifest_list on a FIFO is refused, not a hang (XX001)")
+
+
+def test_read_parquet_refuses_a_fifo_beside_iceberg(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["fifo parquet"], "XX001 (not a regular file)", "read_parquet on a FIFO is refused, not a hang (XX001)")
+
+
+def test_the_backend_survives_the_fifo_refusals(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["fifo up"], "1", "backend still up after the FIFO refusals")
+
+
+def test_a_manifest_list_that_is_a_fifo_is_refused(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["fifo ml"], "XX001 (not a regular file)",
+                "a manifest-list that is a FIFO is refused, not a hang (XX001)")
+
+
+def test_the_backend_survives_the_manifest_list_fifo(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["fifo ml up"], "1", "backend still up after the manifest-list FIFO refusal")
+
+
+def test_a_null_manifest_list_sequence_number_reads_null(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["seq null"], "t", "a null manifest-list sequence_number decodes as NULL, not 0")
+
+
+def test_a_present_manifest_list_sequence_number_reads_its_value(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["seq ctl"], "5", "control: a present manifest-list sequence_number decodes to its value")
+
+
+def test_a_null_min_sequence_number_reads_null(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["minseq null"], "t", "a null manifest-list min_sequence_number decodes as NULL, not 0 (#686)")
+
+
+def test_a_present_min_sequence_number_reads_its_value(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["minseq ctl"], "7", "control: a present min_sequence_number decodes to its value (#686)")
+
+
+def test_the_scan_refuses_a_null_sequence_an_added_entry_inherits(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["seq refused"], "XX001 (null sequence to inherit)",
+                "iceberg_scan refuses a null manifest seq an ADDED entry must inherit (XX001, not a silent wrong row)")
+
+
+def test_the_backend_survives_the_null_sequence_refusal(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["seq up"], "1", "backend still up after the null-sequence refusal")
+
+
+def test_a_concrete_manifest_sequence_scans_every_row(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["seq control"], "5",
+                "control: a concrete manifest seq scans all 5 rows (seq-3 delete is older than seq-5 data)")
+
+
+def test_a_dangling_current_schema_id_is_refused(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["dangling"], "XX001 (dangling schema id)",
+                "a dangling current-schema-id is refused (XX001), not silently stale-read")
+
+
+def test_the_backend_survives_the_dangling_schema_refusal(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["dangling up"], "1", "backend still up after the dangling-schema refusal")
+
+
+def test_a_resolvable_current_schema_id_reads_amount(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["dangling control"], "10,20,30,40,50",
+                "control: a resolvable current-schema-id reads the correct amount column")
+
+
+def test_a_legacy_only_schema_table_still_reads(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["legacy rows"], "5", "control: a legacy-only-schema table still reads (5 rows)")
+
+
+def test_the_legacy_table_binds_amount(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["legacy amount"], "10,20,30,40,50", "control: the legacy table binds amount correctly (10..50)")
diff --git a/test/pytest/unported_suites.txt b/test/pytest/unported_suites.txt
index 94ddacc08..03d9da374 100644
--- a/test/pytest/unported_suites.txt
+++ b/test/pytest/unported_suites.txt
@@ -21,7 +21,6 @@ harness_selftest
 hilbert_curve
 iceberg_deletes
 iceberg_dv
-iceberg_malformed
 iceberg_name_mapping
 iceberg_objstore
 iceberg_rest