143
run_mode: str = "normal",
144
source_run_id: str = "",
145
raw_store_manifest: Path | None = None,
146
+ synthesis_status: str | None = "available",
147
+ synthesis_file: Path | None = None,
148
) -> list[str]:
149
args = [
150
"create",
181
args.extend(["--preflight-report", str(preflight_path)])
182
elif isinstance(preflight, Path):
183
args.extend(["--preflight-report", str(preflight)])
184
+ if synthesis_status is not None:
185
+ args.extend(["--synthesis-status", synthesis_status])
186
+ # When we claim synthesis is available we must back it with real provenance
187
+ # (a readable, non-empty file), matching how the workflow signals "available".
188
+ if synthesis_status == "available" and synthesis_file is None:
189
+ synthesis_file = manifest.parent / "diagnostics" / "synthesis-narrative.md"
190
+ synthesis_file.parent.mkdir(parents=True, exist_ok=True)
191
+ synthesis_file.write_text("Weekly synthesis narrative.\n", encoding="utf-8")
192
+ if synthesis_file is not None:
193
+ args.extend(["--synthesis-file", str(synthesis_file)])
194
if source_run_id:
195
args.extend(["--source-run-id", source_run_id])
196
if raw_store_manifest is not None:
1268
)
1269
1270
1271
+class FailClosedSynthesisTests(unittest.TestCase):
1272
+ """Cover issue #571: required synthesis must fail closed for normal AI publication."""
1273
+
1274
+ def _prepare(self, base: Path) -> tuple[Path, Path, Path, Path]:
1275
+ raw = base / "data/raw/2026-W21.json"
1276
+ summary = base / "data/candidates/2026-W21/123456/2026-W21-summary.md"
1277
+ manifest = base / "data/candidates/2026-W21/123456/publish-manifest.json"
1278
+ gate_report = base / "data/candidates/2026-W21/123456/analysis-gate-report.json"
1279
+ write_raw(raw)
1280
+ write_summary(summary)
1281
+ write_gate_report(gate_report)
1282
+ return raw, summary, manifest, gate_report
1283
+
1284
+ def _assert_synthesis_blocks(self, status: str) -> None:
1285
+ tests_root = Path(__file__).resolve().parent
1286
+ with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
1287
+ base = Path(tmpdir)
1288
+ raw, summary, manifest, gate_report = self._prepare(base)
1289
+
1290
+ publish_manifest.main(
1291
+ create_args(
1292
+ base,
1293
+ raw,
1294
+ summary,
1295
+ manifest,
1296
+ gate_report=gate_report,
1297
+ synthesis_status=status,
1298
+ )
1299
+ )
1300
+
1301
+ payload = json.loads(manifest.read_text(encoding="utf-8"))
1302
+ self.assertFalse(payload["promotion"]["eligible"])
1303
+ self.assertTrue(payload["synthesis"]["required"])
1304
+ self.assertEqual(payload["synthesis"]["status"], status)
1305
+ self.assertFalse(payload["synthesis"]["available"])
1306
+ self.assertTrue(
1307
+ any(
1308
+ f"required synthesis is {status}" in reason
1309
+ for reason in payload["promotion"]["reasons"]
1310
+ ),
1311
+ f"expected synthesis reason for status={status!r} in {payload['promotion']['reasons']}",
1312
+ )
1313
+ with self.assertRaises(SystemExit) as ctx:
1314
+ assert_eligible_from_root(base, manifest)
1315
+ self.assertIn("synthesis", str(ctx.exception).lower())
1316
+ self.assertIn(status, str(ctx.exception))
1317
+
1318
+ def test_missing_synthesis_blocks_normal_ai_publication(self) -> None:
1319
+ self._assert_synthesis_blocks("missing")
1320
+
1321
+ def test_empty_synthesis_blocks_normal_ai_publication(self) -> None:
1322
+ self._assert_synthesis_blocks("empty")
1323
+
1324
+ def test_failed_synthesis_blocks_normal_ai_publication(self) -> None:
1325
+ self._assert_synthesis_blocks("failed")
1326
+
1327
+ def test_available_synthesis_records_provenance_on_manifest(self) -> None:
1328
+ tests_root = Path(__file__).resolve().parent
1329
+ with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
1330
+ base = Path(tmpdir)
1331
+ raw, summary, manifest, gate_report = self._prepare(base)
1332
+ synthesis_file = base / "data/diagnostics/2026-W21/synthesis.md"
1333
+ synthesis_file.parent.mkdir(parents=True, exist_ok=True)
1334
+ synthesis_file.write_text("# Weekly synthesis narrative\n", encoding="utf-8")
1335
+
1336
+ args = create_args(
1337
+ base,
1338
+ raw,
1339
+ summary,
1340
+ manifest,
1341
+ gate_report=gate_report,
1342
+ synthesis_status="available",
1343
+ synthesis_file=synthesis_file,
1344
+ )
1345
+ self.assertEqual(publish_manifest.main(args), 0)
1346
+
1347
+ payload = json.loads(manifest.read_text(encoding="utf-8"))
1348
+ self.assertTrue(payload["promotion"]["eligible"])
1349
+ self.assertTrue(payload["synthesis"]["required"])
1350
+ self.assertEqual(payload["synthesis"]["status"], "available")
1351
+ self.assertTrue(payload["synthesis"]["available"])
1352
+ self.assertEqual(
1353
+ payload["synthesis"]["path"],
1354
+ synthesis_file.as_posix(),
1355
+ )
1356
+ self.assertRegex(payload["synthesis"]["sha256"], r"^[0-9a-f]{64}$")
1357
+ self.assertEqual(assert_eligible_from_root(base, manifest), 0)
1358
+
1359
+ def test_dry_run_mode_is_not_gated_by_synthesis(self) -> None:
1360
+ """Non-normal/debug modes remain isolated from the fail-closed gate."""
1361
+ tests_root = Path(__file__).resolve().parent
1362
+ with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
1363
+ base = Path(tmpdir)
1364
+ raw, summary, manifest, gate_report = self._prepare(base)
1365
+
1366
+ publish_manifest.main(
1367
+ create_args(
1368
+ base,
1369
+ raw,
1370
+ summary,
1371
+ manifest,
1372
+ gate_report=gate_report,
1373
+ synthesis_status="missing",
1374
+ run_mode="dry-run",
1375
+ )
1376
+ )
1377
+
1378
+ payload = json.loads(manifest.read_text(encoding="utf-8"))
1379
+ self.assertFalse(payload["synthesis"]["required"])
1380
+ self.assertFalse(
1381
+ any("required synthesis" in reason for reason in payload["promotion"]["reasons"]),
1382
+ f"dry-run must not emit synthesis reasons: {payload['promotion']['reasons']}",
1383
+ )
1384
+
1385
+ def test_no_ai_normal_mode_is_not_gated_by_synthesis(self) -> None:
1386
+ """no-ai fallback publication is governed by its own force-replace path, not synthesis."""
1387
+ tests_root = Path(__file__).resolve().parent
1388
+ with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
1389
+ base = Path(tmpdir)
1390
+ raw, summary, manifest, gate_report = self._prepare(base)
1391
+
1392
+ publish_manifest.main(
1393
+ create_args(
1394
+ base,
1395
+ raw,
1396
+ summary,
1397
+ manifest,
1398
+ source="no-ai",
1399
+ model="none",
1400
+ gate_report=gate_report,
1401
+ synthesis_status="missing",
1402
+ )
1403
+ )
1404
+
1405
+ payload = json.loads(manifest.read_text(encoding="utf-8"))
1406
+ self.assertFalse(payload["synthesis"]["required"])
1407
+ self.assertFalse(
1408
+ any("required synthesis" in reason for reason in payload["promotion"]["reasons"]),
1409
+ f"no-ai mode must not emit synthesis reasons: {payload['promotion']['reasons']}",
1410
+ )
1411
+
1412
+ def test_available_without_file_is_downgraded_and_fails_closed(self) -> None:
1413
+ """Claiming 'available' without provenance downgrades to missing and blocks promotion."""
1414
+ tests_root = Path(__file__).resolve().parent
1415
+ with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
1416
+ base = Path(tmpdir)
1417
+ raw, summary, manifest, gate_report = self._prepare(base)
1418
+
1419
+ # Pass status "available" but no synthesis file / provenance.
1420
+ args = create_args(
1421
+ base,
1422
+ raw,
1423
+ summary,
1424
+ manifest,
1425
+ gate_report=gate_report,
1426
+ synthesis_status=None,
1427
+ )
1428
+ args.extend(["--synthesis-status", "available"])
1429
+ self.assertEqual(publish_manifest.main(args), 0)
1430
+
1431
+ payload = json.loads(manifest.read_text(encoding="utf-8"))
1432
+ self.assertTrue(payload["synthesis"]["required"])
1433
+ self.assertEqual(payload["synthesis"]["status"], "missing")
1434
+ self.assertFalse(payload["synthesis"]["available"])
1435
+ self.assertIsNone(payload["synthesis"]["path"])
1436
+ self.assertIsNone(payload["synthesis"]["sha256"])
1437
+ self.assertFalse(payload["promotion"]["eligible"])
1438
+ self.assertTrue(
1439
+ any(
1440
+ "no readable, non-empty" in reason for reason in payload["synthesis"]["reasons"]
1441
+ ),
1442
+ payload["synthesis"]["reasons"],
1443
+ )
1444
+
1445
+ def test_available_with_empty_file_is_downgraded(self) -> None:
1446
+ """An empty synthesis file cannot back an 'available' claim."""
1447
+ tests_root = Path(__file__).resolve().parent
1448
+ with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
1449
+ base = Path(tmpdir)
1450
+ raw, summary, manifest, gate_report = self._prepare(base)
1451
+ empty = base / "diagnostics" / "synthesis-narrative.md"
1452
+ empty.parent.mkdir(parents=True, exist_ok=True)
1453
+ empty.write_text("", encoding="utf-8")
1454
+
1455
+ args = create_args(
1456
+ base,
1457
+ raw,
1458
+ summary,
1459
+ manifest,
1460
+ gate_report=gate_report,
1461
+ synthesis_status="available",
1462
+ synthesis_file=empty,
1463
+ )
1464
+ self.assertEqual(publish_manifest.main(args), 0)
1465
+
1466
+ payload = json.loads(manifest.read_text(encoding="utf-8"))
1467
+ self.assertEqual(payload["synthesis"]["status"], "missing")
1468
+ self.assertIsNone(payload["synthesis"]["path"])
1469
+ self.assertFalse(payload["promotion"]["eligible"])
1470
+
1471
+ def test_assert_eligible_requires_synthesis_provenance(self) -> None:
1472
+ """assert-eligible rejects a manifest that claims available without path/sha256."""
1473
+ tests_root = Path(__file__).resolve().parent
1474
+ with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
1475
+ base = Path(tmpdir)
1476
+ raw, summary, manifest, gate_report = self._prepare(base)
1477
+
1478
+ self.assertEqual(
1479
+ publish_manifest.main(
1480
+ create_args(base, raw, summary, manifest, gate_report=gate_report)
1481
+ ),
1482
+ 0,
1483
+ )
1484
+ payload = json.loads(manifest.read_text(encoding="utf-8"))
1485
+ # Tamper: strip provenance while leaving status "available".
1486
+ payload["synthesis"]["path"] = None
1487
+ payload["synthesis"]["sha256"] = None
1488
+ manifest.write_text(json.dumps(payload), encoding="utf-8")
1489
+
1490
+ with self.assertRaises(SystemExit) as ctx:
1491
+ assert_eligible_from_root(base, manifest)
1492
+ self.assertIn("provenance", str(ctx.exception))
1493
+
1494
+
1495
if __name__ == "__main__":
1496
unittest.main()