344 lines
23 KiB
Python
344 lines
23 KiB
Python
"""Offline r26 harness contract: no auth, bootstrap, socket or model dispatch."""
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
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_r26_harness", OUT / "private_tcm_r26_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
|
|
|
|
|
|
def lifecycle_summary(status="stopped", *, captured=True):
|
|
terminal = captured and status in {"stopped", "succeeded", "failed"}
|
|
snapshot = 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=20,
|
|
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 captured 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 captured),
|
|
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=1, original_run_count=int(captured),
|
|
audit_bytes=2048, audit_sha256="a" * 64, closed_generation_count=1,
|
|
independently_terminal_generation_count=int(terminal), pending_generation_count=int(not terminal),
|
|
unidentified_generation_count=int(not captured), 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=True, submission=dict(submitted_at=8.0, transfer_closed=True, transfer_closed_at=11.0, disposition="unconfirmed"), original_pair_captured=captured, transfer_closed=True, closed_at_request_ms=5.0,
|
|
transfer_exit="natural_eof", independently_terminal=terminal, lifecycle=snapshot)])
|
|
|
|
|
|
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 PrivateTcmR26HarnessTest(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_r26_wire_audit.php"))
|
|
|
|
def test_auth_preparation_delivery_and_export_remain_original(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 =")], "Authentication, permissions and caller override limits unchanged")
|
|
for begin, end in [(" // No real preparation", " $testedHashes="),
|
|
(" $reply = ", " if ($reply === null)"),
|
|
(" $text=(string)", " $row=")]:
|
|
self.assertEqual(old[old.index(begin):old.index(end)],new[new.index(begin):new.index(end)])
|
|
self.assertIn("closeGeneration($taskId, [\\app\\common\\service\\DifyWorkflowRunStatus::class, 'lookup'])",new)
|
|
self.assertNotIn("clearTask($taskId !== '' ? $taskId : null)",new)
|
|
self.assertLess(new.index("$snapshot = $timing->snapshot();"),new.index("$privateWireAudit->transferClosedCurrent("))
|
|
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")
|
|
marker = ' if result.get("preparation_isolation_verified") is True and save_body:'
|
|
self.assertEqual(old[old.index(marker):old.index(" return result",old.index(marker))],new[new.index(marker):new.index(" return result",new.index(marker))])
|
|
|
|
def test_safe_summary_has_strict_types_and_no_identity_channel(self):
|
|
valid = lifecycle_summary()
|
|
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": 2}, {**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_r26_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"], "v43r26_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_r26_wire_audit.php"),
|
|
},
|
|
"harness_contract": "v43r26_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": lifecycle_summary(),
|
|
"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 with_counts(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"] = lifecycle_summary(captured=False)
|
|
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")
|
|
|
|
|
|
def test_running_footer_is_durable_but_never_independent_cleanup(self):
|
|
summary = lifecycle_summary("running")
|
|
self.assertEqual(driver.safe_private_wire_summary(summary), summary)
|
|
row = dict(preparation_isolation_verified=True, private_wire_audit=summary,
|
|
state=dict(transport_closed=True, upstream_terminal_confirmed=True, upstream_pending_task_count=0))
|
|
self.assertFalse(driver.independent_cleanup_confirmed(with_counts(row)), "a false terminal public flag cannot override running observations")
|
|
row["private_wire_audit"] = lifecycle_summary()
|
|
self.assertTrue(driver.independent_cleanup_confirmed(with_counts(row)))
|
|
row["state"]["upstream_pending_task_count"] = True
|
|
self.assertFalse(driver.independent_cleanup_confirmed(with_counts(row)), "boolean is not a task count")
|
|
|
|
def test_false_terminal_foreign_fields_and_budget_mutations_fail_closed(self):
|
|
controls = [
|
|
lambda a: a["generations"][0].update(original_task_id="must-stay-private"),
|
|
lambda a: a["generations"][0]["lifecycle"].update(original_workflow_run_id="must-stay-private"),
|
|
lambda a: a["generations"][0]["lifecycle"]["observations"][0].update(task_id="must-stay-private"),
|
|
lambda a: a["generations"][0]["lifecycle"]["observations"][0].update(within_budget=False),
|
|
lambda a: a["generations"][0].update(transfer_closed=False),
|
|
lambda a: a["generations"][0].update(closed_at_request_ms=float("nan")),
|
|
lambda a: a["generations"][0]["lifecycle"].update(persisted_run_status="running"),
|
|
lambda a: a["generations"][0]["lifecycle"].update(verification_attempts=3),
|
|
lambda a: a["generations"][0]["lifecycle"].update(verification_attempts=True),
|
|
lambda a: a["generations"][0]["lifecycle"].update(persisted_run_observed_at=13.0),
|
|
lambda a: a["generations"][0].update(generation=2),
|
|
lambda a: a["generations"][0].update(transfer_exit={"foreign": "value"}),
|
|
lambda a: a["generations"][0]["lifecycle"].update(persisted_run_status=[]),
|
|
lambda a: a["generations"][0]["lifecycle"].update(producer_terminal_observed_at=None),
|
|
lambda a: a["generations"][0]["lifecycle"].update(verification_started_at=None),
|
|
]
|
|
for change in controls:
|
|
value = lifecycle_summary(); change(value)
|
|
self.assertEqual(driver.safe_private_wire_summary(value), {"persisted": False})
|
|
running = lifecycle_summary("running")
|
|
running["independent_cleanup_confirmed"] = True
|
|
self.assertEqual(driver.safe_private_wire_summary(running), {"persisted": False})
|
|
|
|
def test_deadline_sequence_blocks_all_undispatched_slots_after_pending(self):
|
|
writes = []; actual_open = Path.open
|
|
@contextlib.contextmanager
|
|
def memory_open(path, mode="r", *args, **kwargs):
|
|
if mode == "x":
|
|
stream = io.StringIO(); yield stream; writes.append(json.loads(stream.getvalue())); stream.close()
|
|
else:
|
|
with actual_open(path, mode, *args, **kwargs) as stream:
|
|
yield stream
|
|
pending = dict(preparation_isolation_verified=True, production_acceptance_path=True, diagnostic_only=False,
|
|
ok=False, error_code="UPSTREAM_TIMEOUT", private_wire_audit=lifecycle_summary("running"),
|
|
state=dict(cancel_requested=False, timeout_reached=True, transport_closed=True,
|
|
upstream_terminal_confirmed=True, upstream_stop_confirmed=True, upstream_pending_task_count=0))
|
|
with patch.object(driver, "run", return_value=pending) as dispatch, \
|
|
patch.object(driver, "_local_fcgi", side_effect=AssertionError("no auth")) 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("offline_pending", ["qwen", "openai"], driver.argparse.ArgumentParser())
|
|
self.assertEqual(dispatch.call_count, 1)
|
|
auth.assert_not_called()
|
|
final = writes[-1]
|
|
self.assertEqual((final["completed_requests"], len(final["rows"]), len(final["unexecuted_requests"])), (1, 1, 3))
|
|
self.assertFalse(final["suite_complete"])
|
|
self.assertFalse(final["rows"][0]["ordinary_deadline_pass"])
|
|
self.assertEqual({row["reason"] for row in final["unexecuted_requests"]}, {"PREDECESSOR_CLEANUP_UNCONFIRMED"})
|
|
|
|
def test_deadline_sequence_allows_next_full_only_after_independent_terminal(self):
|
|
writes = []; actual_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 actual_open(path, mode, *args, **kwargs) as stream:
|
|
yield stream
|
|
def reply(profile, label, **options):
|
|
controlled = options["controlled_deadline"]
|
|
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=lifecycle_summary(),
|
|
state=dict(cancel_requested=False, timeout_reached=controlled, transport_closed=True,
|
|
upstream_terminal_confirmed=True, upstream_stop_confirmed=True, upstream_pending_task_count=0)))
|
|
with patch.object(driver, "run", side_effect=reply) as dispatch, \
|
|
patch.object(driver, "_local_fcgi", side_effect=AssertionError("no auth")) 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("offline_clean", ["qwen"], driver.argparse.ArgumentParser())
|
|
auth.assert_not_called()
|
|
self.assertEqual(dispatch.call_count, 2)
|
|
final = writes[-1]
|
|
self.assertEqual(final["completed_requests"], 2)
|
|
self.assertTrue(final["suite_complete"])
|
|
self.assertTrue(final["rows"][0]["ordinary_deadline_pass"])
|
|
self.assertTrue(final["rows"][1]["next_full_production_acceptance_pass"])
|
|
self.assertEqual(final["unexecuted_requests"], [])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|