224 lines
14 KiB
Python
224 lines
14 KiB
Python
"""Offline r25 harness contract: no auth, bootstrap, socket or model dispatch."""
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import re
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
OUT = Path(__file__).resolve().parents[2] / "app/artifacts/ai_concurrency"
|
|
spec = importlib.util.spec_from_file_location("private_r25_harness", OUT / "private_tcm_r25_acceptance_driver.py")
|
|
driver = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(driver)
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def no_auth_or_result_writes():
|
|
writes = []
|
|
original_open = Path.open
|
|
|
|
def guarded_open(path, mode="r", *args, **kwargs):
|
|
if any(flag in mode for flag in "wax+"):
|
|
writes.append((str(path), mode))
|
|
raise AssertionError("result file must not open for writing")
|
|
return original_open(path, mode, *args, **kwargs)
|
|
|
|
with patch.object(driver, "_local_fcgi", side_effect=AssertionError("auth must not load")) as auth, \
|
|
patch.object(driver, "uuid4", side_effect=AssertionError("rejected label must not allocate a request")) as identity, \
|
|
patch.object(Path, "open", new=guarded_open), \
|
|
patch.object(Path, "write_text", side_effect=AssertionError("result file must not be written")) as writer:
|
|
yield auth, identity, writer, writes
|
|
|
|
|
|
class PrivateTcmR25HarnessTest(unittest.TestCase):
|
|
def test_historical_and_fixed_source_identities(self):
|
|
identities = driver.validate_local_sources()
|
|
self.assertEqual(identities["fixture_sha256"], driver.IMMUTABLE_FIXTURE_SHA256)
|
|
self.assertEqual(identities["canonical_question_sha256"], driver.IMMUTABLE_QUESTION_SHA256)
|
|
self.assertEqual(len(identities["probe_hashes"]), 2)
|
|
self.assertEqual(identities["private_wire_audit_helper_sha256"], driver.sha256(OUT / "private_tcm_r25_wire_audit.php"))
|
|
|
|
def test_probe_delta_is_exclusively_private_observability(self):
|
|
for kind in ("acceptance", "deadline"):
|
|
old = (OUT / f"private_tcm_r24_{kind}_probe.php").read_text(encoding="utf-8")
|
|
new = (OUT / f"private_tcm_r25_{kind}_probe.php").read_text(encoding="utf-8")
|
|
rebuilt = new.replace("PRIVATE r25 exact private-wire run-aware", "PRIVATE r24 run-aware")
|
|
rebuilt = rebuilt.replace("'v43r25_private_run_wire'", "'v43r24_run_aware'").replace("'/tcm_r25_acceptance_controls'", "'/tcm_r24_acceptance_controls'")
|
|
additions = [
|
|
" $requestLabel = $body['request_label'] ?? '';\n"
|
|
" if ($action === 'run' && (!is_string($requestLabel) || preg_match('/\\A[A-Za-z0-9][A-Za-z0-9_-]{0,100}\\z/D', $requestLabel) !== 1))\n"
|
|
" inventoryFinish(['ok'=>false,'error_code'=>'request_label_required'],400);\n",
|
|
" require_once __DIR__ . '/private_tcm_r25_wire_audit.php';\n"
|
|
" $privateWireAudit = new PrivateTcmR25WireAudit($serverRoot . '/runtime/tcm-r25-wire-audits', $identity, $requestLabel, $profile);\n",
|
|
" $testedHashes['private_wire_audit_helper'] = hash_file('sha256', __DIR__ . '/private_tcm_r25_wire_audit.php');\n",
|
|
" 'on_private_wire_request' => [$privateWireAudit, 'observe'],\n",
|
|
" $privateWireSummary = $privateWireAudit->finish($recoveryAfterState);\n",
|
|
" 'private_wire_audit'=>$privateWireSummary,\n",
|
|
]
|
|
for addition in additions:
|
|
self.assertEqual(rebuilt.count(addition), 1)
|
|
rebuilt = rebuilt.replace(addition, "", 1)
|
|
rebuilt = rebuilt.replace("'request_id','request_label','review'", "'request_id','review'")
|
|
self.assertEqual(rebuilt, old, "every original preparation, generation, control, timeout and export line is preserved")
|
|
self.assertLess(new.index("reconcileClosedTasks("), new.index("$privateWireAudit->finish("))
|
|
|
|
def test_safe_summary_has_strict_types_and_no_identity_channel(self):
|
|
valid = {"persisted": True, "serialized_request_count": 2, "original_task_count": 2,
|
|
"original_run_count": 2, "audit_bytes": 2048, "audit_sha256": "a" * 64}
|
|
self.assertEqual(driver.safe_private_wire_summary(valid), valid)
|
|
for bad in (None, [], {**valid, "original_task_id": "must-stay-private"},
|
|
{**valid, "audit_sha256": "must-stay-private"}, {**valid, "audit_bytes": True},
|
|
{**valid, "original_run_count": 3}, {**valid, "serialized_request_count": 17},
|
|
{**valid, "audit_bytes": 65537}, {**valid, "persisted": "true"}):
|
|
self.assertEqual(driver.safe_private_wire_summary(bad), {"persisted": False})
|
|
|
|
def test_dry_runs_never_load_auth_or_dispatch(self):
|
|
for extra in ([], ["--deadline-sequence"]):
|
|
result = io.StringIO()
|
|
with patch.object(sys, "argv", ["private_tcm_r25_acceptance_driver.py", "--stage", "offline_test", "--profiles", "qwen", "openai", "--dry-run", *extra]), \
|
|
patch.object(driver, "_local_fcgi", side_effect=AssertionError("auth must not load")), contextlib.redirect_stdout(result):
|
|
driver.main()
|
|
document = json.loads(result.getvalue())
|
|
self.assertFalse(document["dispatch"])
|
|
self.assertFalse(document["auth_loaded"])
|
|
self.assertFalse(document["result_files_written"])
|
|
self.assertEqual(document["harness_contract"], "v43r25_private_run_wire")
|
|
self.assertEqual([row["generation_budget_seconds"] for row in document["expected_requests"]], [1, 90, 1, 90] if extra else [90, 90])
|
|
|
|
def test_complete_request_labels_allow_101_reject_102_for_every_kind(self):
|
|
for profile in ("qwen", "openai"):
|
|
for kind in ("normal", "deadline1", "next_full90"):
|
|
suffix = "_0" if kind == "normal" else f"_{profile}_{kind}"
|
|
label = "s" * (101 - len(suffix) - len(profile) - 1) + suffix
|
|
with self.subTest(profile=profile, kind=kind):
|
|
exact = f"{label}_{profile}"
|
|
self.assertEqual(len(exact), 101)
|
|
self.assertEqual(driver.validated_request_label(label, profile), exact)
|
|
driver.validate_manifest_request_labels([{"test_label": label, "profile": profile}])
|
|
with self.assertRaises(ValueError):
|
|
driver.validated_request_label("s" + label, profile)
|
|
with self.assertRaises(ValueError):
|
|
driver.validate_manifest_request_labels([{"test_label": "s" + label, "profile": profile}])
|
|
for invalid in ("bad/label", "bad\nlabel", "中文", "", None):
|
|
with self.subTest(invalid=invalid), self.assertRaises(ValueError):
|
|
driver.validated_request_label(invalid, "qwen")
|
|
|
|
def test_rejected_direct_labels_never_authenticate_allocate_or_write(self):
|
|
with no_auth_or_result_writes() as (auth, identity, writer, writes):
|
|
for profile in ("qwen", "openai"):
|
|
for kind in ("normal", "deadline1", "next_full90"):
|
|
suffix = "_0" if kind == "normal" else f"_{profile}_{kind}"
|
|
label = "s" * (102 - len(suffix) - len(profile) - 1) + suffix
|
|
controlled = kind == "deadline1"
|
|
row = driver.run(profile, label, controlled_deadline=controlled,
|
|
probe_name=driver.DEADLINE_PROBE if controlled else driver.DEFAULT_PROBE)
|
|
self.assertFalse(row["ok"])
|
|
self.assertEqual(row["error_code"], "PRIVATE_DRIVER_REQUEST_FAILED")
|
|
self.assertTrue(row["clinical_body_not_saved"])
|
|
self.assertNotIn("synthetic_review_file", row)
|
|
auth.assert_not_called()
|
|
identity.assert_not_called()
|
|
writer.assert_not_called()
|
|
self.assertEqual(writes, [])
|
|
|
|
def test_deadline_cli_long_label_rejected_before_auth_manifest_or_dry_run(self):
|
|
for dry in ([], ["--dry-run"]):
|
|
output, errors = io.StringIO(), io.StringIO()
|
|
with no_auth_or_result_writes() as (auth, identity, writer, writes), \
|
|
patch.object(sys, "argv", ["driver", "--stage", "s" * 81, "--profiles", "qwen", "openai", "--deadline-sequence", *dry]), \
|
|
contextlib.redirect_stdout(output), contextlib.redirect_stderr(errors):
|
|
with self.assertRaises(SystemExit) as stopped:
|
|
driver.main()
|
|
self.assertEqual(stopped.exception.code, 2)
|
|
auth.assert_not_called()
|
|
identity.assert_not_called()
|
|
writer.assert_not_called()
|
|
self.assertEqual(writes, [])
|
|
self.assertEqual(output.getvalue(), "")
|
|
self.assertIn("generated private request labels", errors.getvalue())
|
|
|
|
def test_deadline_complete_manifest_101_passes_102_rejects_without_io(self):
|
|
for profile, stage_length in (("qwen", 79), ("openai", 75)):
|
|
for overflow in (0, 1):
|
|
output, errors = io.StringIO(), io.StringIO()
|
|
with no_auth_or_result_writes() as (auth, identity, writer, writes), \
|
|
patch.object(sys, "argv", ["driver", "--stage", "s" * (stage_length + overflow), "--profiles", profile, "--deadline-sequence", "--dry-run"]), \
|
|
contextlib.redirect_stdout(output), contextlib.redirect_stderr(errors):
|
|
if overflow:
|
|
with self.assertRaises(SystemExit) as stopped:
|
|
driver.main()
|
|
self.assertEqual(stopped.exception.code, 2)
|
|
else:
|
|
driver.main()
|
|
document = json.loads(output.getvalue())
|
|
labels = [driver.validated_request_label(row["test_label"], row["profile"]) for row in document["expected_requests"]]
|
|
self.assertEqual(max(map(len, labels)), 101)
|
|
self.assertFalse(document["dispatch"])
|
|
self.assertFalse(document["auth_loaded"])
|
|
self.assertFalse(document["result_files_written"])
|
|
auth.assert_not_called()
|
|
identity.assert_not_called()
|
|
writer.assert_not_called()
|
|
self.assertEqual(writes, [])
|
|
|
|
def test_direct_deadline_sequence_checks_all_labels_before_manifest_write(self):
|
|
with no_auth_or_result_writes() as (auth, identity, writer, writes), contextlib.redirect_stderr(io.StringIO()):
|
|
with self.assertRaises(SystemExit) as stopped:
|
|
driver.run_deadline_sequence("s" * 81, ["qwen", "openai"], driver.argparse.ArgumentParser())
|
|
self.assertEqual(stopped.exception.code, 2)
|
|
auth.assert_not_called()
|
|
identity.assert_not_called()
|
|
writer.assert_not_called()
|
|
self.assertEqual(writes, [])
|
|
|
|
def test_mocked_driver_preserves_exact_label_and_private_summary_only(self):
|
|
calls = []
|
|
synthetic = {
|
|
"production_hashes": {
|
|
"fixture": driver.sha256(driver.FIXTURE), "probe": driver.sha256(OUT / driver.DEFAULT_PROBE),
|
|
"original_acceptance_probe": driver.sha256(driver.ORIGINAL_PROBE), "question": driver.canonical_question_hash(),
|
|
"request_control": driver.sha256(driver.ROOT / "server/app/adminapi/service/AssistantRequestControl.php"),
|
|
"status_adapter": driver.sha256(driver.ROOT / "server/app/common/service/DifyWorkflowRunStatus.php"),
|
|
"private_wire_audit_helper": driver.sha256(OUT / "private_tcm_r25_wire_audit.php"),
|
|
},
|
|
"harness_contract": "v43r25_private_run_wire", "terminal_recovery_before_state": {}, "state": {},
|
|
"terminal_recovery_after_state": {}, "terminal_recovery_elapsed_ms": 0,
|
|
"terminal_recovery_outcome": {"status": "complete"},
|
|
"synthetic": True, "preparation_isolated": True, "production_acceptance_path": True, "diagnostic_only": False,
|
|
"profile": "qwen", "case_id": 11, "mode": "chat", "attachment_count": 0, "guard_restricted": False,
|
|
"requested_report_char_range": [1800, 2200], "effective_reference_scope": "case", "deadline_probe": False,
|
|
"generation_budget_seconds": 90, "ok": True,
|
|
"private_wire_audit": {"persisted": True, "serialized_request_count": 1, "original_task_count": 1,
|
|
"original_run_count": 1, "audit_bytes": 1024, "audit_sha256": "b" * 64},
|
|
"original_task_id": "must-stay-private-task", "original_workflow_run_id": "must-stay-private-run",
|
|
}
|
|
def call(payload, probe):
|
|
calls.append((payload.copy(), probe))
|
|
return dict(synthetic)
|
|
fake = SimpleNamespace(token=lambda: "synthetic-never-authenticated", call=call)
|
|
with patch.object(driver, "_local_fcgi", return_value=fake):
|
|
row = driver.run("qwen", "offline_exact_label", save_body=False)
|
|
self.assertTrue(row["preparation_isolation_verified"])
|
|
self.assertTrue(row["private_wire_audit_verified"])
|
|
self.assertEqual(calls[0][0]["request_label"], "offline_exact_label_qwen")
|
|
self.assertEqual(calls[0][1], driver.DEFAULT_PROBE)
|
|
self.assertRegex(calls[0][0]["request_id"], r"^[0-9a-f-]{36}$")
|
|
self.assertNotIn("must-stay-private", json.dumps(row))
|
|
self.assertNotIn("synthetic-never-authenticated", json.dumps(row))
|
|
self.assertNotIn(calls[0][0]["request_id"], json.dumps(row))
|
|
synthetic["private_wire_audit"] = {**synthetic["private_wire_audit"], "original_run_count": 0}
|
|
with patch.object(driver, "_local_fcgi", return_value=fake):
|
|
incomplete = driver.run("qwen", "offline_missing_run", save_body=False)
|
|
self.assertTrue(incomplete["private_wire_audit"]["persisted"])
|
|
self.assertFalse(incomplete["private_wire_audit_verified"], "retention alone cannot pretend that the original run was captured")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|