"""Pure immutable r26 driver/mapper review. No auth or production import.""" import ast import contextlib import copy import importlib.util import io import json from pathlib import Path import unittest from unittest.mock import patch OUT = Path(__file__).resolve().parents[2] / "app/artifacts/ai_concurrency" spec = importlib.util.spec_from_file_location("r26_driver_independent", OUT / "private_tcm_r26_acceptance_driver.py") driver = importlib.util.module_from_spec(spec) spec.loader.exec_module(driver) def summary(status="stopped", *, identified=True): terminal = identified and status in {"stopped", "succeeded", "failed"} lifecycle = dict(producer_terminal_status="stopped", persisted_run_status=status, independently_terminal=terminal, generation_transfer_closed=True, lifecycle_conflict=status != "stopped", stop_accepted=True, verification_attempts=1, verification_elapsed_ms=25, observations=[dict(status=status, observed_at=12.0, within_budget=True)], producer_terminal_observed_at=10.0, persisted_run_observed_at=12.0, stop_accepted_at=9.0, transfer_closed_at=11.0, verification_started_at=12345.0, clock_domains=dict(wall="unix_seconds_microtime", verification_started_at="monotonic_seconds_hrtime", verification_elapsed_ms="monotonic_milliseconds")) if identified else None return dict(audit_schema="v43r26b_submissions", generation_count=1, submitted_attempt_count=1, submitted_transfer_closed_count=1, submitted_pending_attempt_count=int(not terminal), submitted_without_identity_count=int(not identified), pre_execution_closed_attempt_count=0, not_submitted_attempt_count=0, rejected_before_execution_attempt_count=0, never_submitted_generation_count=0, no_execution_generation_count=0, persisted=True, serialized_request_count=1, original_task_count=int(identified), original_run_count=int(identified), audit_bytes=1200, audit_sha256="b" * 64, closed_generation_count=1, independently_terminal_generation_count=int(terminal), pending_generation_count=int(not terminal), unidentified_generation_count=int(not identified), lifecycle_consistent=True, independent_cleanup_confirmed=terminal, generations=[dict(generation=1, serialized_request=True, submitted=True, not_submitted=False, no_execution_confirmed=False, original_task_captured=identified, submission=dict(submitted_at=8.0, transfer_closed=True, transfer_closed_at=11.0, disposition="unconfirmed"), original_pair_captured=identified, transfer_closed=True, closed_at_request_ms=10.0, transfer_exit="natural_eof", independently_terminal=terminal, lifecycle=lifecycle)]) def result(status="stopped", *, identified=True, controlled=False): return with_counts(dict(preparation_isolation_verified=True, production_acceptance_path=True, diagnostic_only=False, ok=not controlled, error_code="UPSTREAM_TIMEOUT" if controlled else "", private_wire_audit=summary(status, identified=identified), state=dict(transport_closed=True, upstream_terminal_confirmed=identified and status != "running", upstream_stop_confirmed=True, cancel_requested=False, timeout_reached=controlled, upstream_pending_task_count=int(not identified or status == "running")))) def with_counts(row): # Derivative fixtures supply the canonical new Control counters; assertions # below still test all original pending/ownership/denominator outcomes. row = copy.deepcopy(row) audit = row["private_wire_audit"] for key in ("submitted_attempt_count", "submitted_transfer_closed_count", "submitted_pending_attempt_count", "submitted_without_identity_count", "pre_execution_closed_attempt_count", "not_submitted_attempt_count", "rejected_before_execution_attempt_count"): row["state"][key] = audit[key] row["state"].setdefault("upstream_pending_task_count", audit["pending_generation_count"]) if "terminal_recovery_after_state" in row: row["terminal_recovery_after_state"] = copy.deepcopy(row["state"]) return row class IndependentDriver(unittest.TestCase): def test_fully_unknown_generation_is_durable_but_never_clean(self): value = summary(identified=False) self.assertEqual(driver.safe_private_wire_summary(value), value) row = result(identified=False) row["state"].update(upstream_terminal_confirmed=True, upstream_pending_task_count=0) self.assertFalse(driver.independent_cleanup_confirmed(with_counts(row))) self.assertEqual(value["pending_generation_count"], 1) def test_safe_summary_rejects_raw_nested_fields_and_false_terminal(self): for status in ("running", "paused", "partial-succeeded"): value = summary(status) self.assertEqual(driver.safe_private_wire_summary(value), value) value["independent_cleanup_confirmed"] = True self.assertEqual(driver.safe_private_wire_summary(value), {"persisted": False}) for path in ([], ["generations", 0], ["generations", 0, "lifecycle"], ["generations", 0, "lifecycle", "observations", 0]): value = summary(); target = value for key in path: target = target[key] target["original_task_id"] = "private-identity-must-not-export" self.assertEqual(driver.safe_private_wire_summary(value), {"persisted": False}) def test_full90_pending_holds_remaining_profile_slots(self): writes = []; original_open = Path.open @contextlib.contextmanager def memory_open(path, mode="r", *args, **kwargs): if mode == "x": stream = io.StringIO(); yield stream; stream.close() else: with original_open(path, mode, *args, **kwargs) as stream: yield stream replies = [result(controlled=True), result("running")] with patch.object(driver, "run", side_effect=replies) as dispatch, \ patch.object(driver, "_local_fcgi", side_effect=AssertionError("auth forbidden")) as auth, \ patch.object(Path, "exists", return_value=False), patch.object(Path, "open", new=memory_open), \ patch.object(Path, "write_text", side_effect=lambda text, **kw: writes.append(json.loads(text))), \ contextlib.redirect_stdout(io.StringIO()): driver.run_deadline_sequence("independent_full_pending", ["qwen", "openai"], driver.argparse.ArgumentParser()) self.assertEqual(dispatch.call_count, 2) auth.assert_not_called() final = writes[-1] self.assertEqual((len(final["expected_requests"]), final["completed_requests"], len(final["unexecuted_requests"])), (4, 2, 2)) self.assertFalse(final["suite_complete"]) self.assertTrue(final["rows"][0]["ordinary_deadline_pass"]) self.assertFalse(final["rows"][1]["next_full_production_acceptance_pass"]) self.assertEqual({row["reason"] for row in final["unexecuted_requests"]}, {"PREDECESSOR_CLEANUP_UNCONFIRMED"}) def test_unknown_deadline_predecessor_never_dispatches_full90(self): writes = []; original_open = Path.open @contextlib.contextmanager def memory_open(path, mode="r", *args, **kwargs): if mode == "x": stream = io.StringIO(); yield stream; stream.close() else: with original_open(path, mode, *args, **kwargs) as stream: yield stream with patch.object(driver, "run", return_value=result(identified=False, controlled=True)) as dispatch, \ patch.object(driver, "_local_fcgi", side_effect=AssertionError("auth forbidden")) as auth, \ patch.object(Path, "exists", return_value=False), patch.object(Path, "open", new=memory_open), \ patch.object(Path, "write_text", side_effect=lambda text, **kw: writes.append(json.loads(text))), \ contextlib.redirect_stdout(io.StringIO()): driver.run_deadline_sequence("independent_unknown", ["qwen"], driver.argparse.ArgumentParser()) self.assertEqual(dispatch.call_count, 1); auth.assert_not_called() self.assertEqual(writes[-1]["completed_requests"], 1) self.assertEqual(len(writes[-1]["unexecuted_requests"]), 1) self.assertFalse(writes[-1]["rows"][0]["ordinary_deadline_pass"]) def test_export_auth_question_and_clinical_callback_preserved(self): for kind in ("acceptance", "deadline"): old = (OUT / f"private_tcm_r25_{kind}_probe.php").read_text(encoding="utf-8") new = (OUT / f"private_tcm_r26_{kind}_probe.php").read_text(encoding="utf-8").replace("r26", "r25").replace("R26", "R25") self.assertEqual(old[:old.index(" $controlDir =")], new[:new.index(" $controlDir =")]) for start, end in [(" // No real preparation", " $testedHashes="), (" $reply = ", " if ($reply === null)")]: self.assertEqual(old[old.index(start):old.index(end)], new[new.index(start):new.index(end)]) self.assertLess(new.index("$control->markTerminal($taskId,$status,$workflowRunId)"), new.index("$privateWireAudit->captureControl($taskId)", new.index("'on_terminal'"))) self.assertIn("$control->closeGeneration($taskId, [\\app\\common\\service\\DifyWorkflowRunStatus::class, 'lookup'])", new) old = (OUT / "private_tcm_r25_acceptance_driver.py").read_text(encoding="utf-8") new = (OUT / "private_tcm_r26_acceptance_driver.py").read_text(encoding="utf-8") start = ' if result.get("preparation_isolation_verified") is True and save_body:' self.assertEqual(old[old.index(start):old.index(" return result", old.index(start))], new[new.index(start):new.index(" return result", new.index(start))]) def test_mapper_is_exact_revision_and_pin_transformation(self): old = (OUT / "clinical_ttft_r25_review.py").read_text(encoding="utf-8") new = (OUT / "clinical_ttft_r26_review.py").read_text(encoding="utf-8") new = new.replace("9c264a7f2871e62998b1328d2e892e35d8038a976120c591705ca887e83dcc10", "263a1a10fa0285dabffe26bb05ebc9683919e7f6b50552949a52d7ed2205e849") new = new.replace("80e8eb0b4feac04d4c5d6837e89e9a4c67942a257986bbb3c5e19bd87678773f", "8858793fe6f189782749345867613a8aa0d9a8f622089173448034874b87a0de") new = new.replace("r26", "r25").replace("R26", "R25") self.assertEqual(ast.dump(ast.parse(old)), ast.dump(ast.parse(new))) if __name__ == "__main__": unittest.main()