158 lines
9.7 KiB
Python
158 lines
9.7 KiB
Python
"""Case-local async test handoff; external delivery and authentication are adapters."""
|
|
|
|
import hashlib
|
|
from viano_bug_analysis.telemetry.decorators import observed
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
from viano_bug_analysis.budget import BudgetManager, BudgetError
|
|
from viano_bug_analysis.case.contracts import read_json, require_file, validate
|
|
from viano_bug_analysis.case.errors import CaseBuildError
|
|
from viano_bug_analysis.package import PackageReader, PackageReadError
|
|
from viano_bug_analysis.package.validator import ARCHIVE_ERRORS
|
|
|
|
from .checkpoints import checkpoint_chain
|
|
from .errors import LifecycleError, LockBusy
|
|
from .locking import FileLock
|
|
from .state import State
|
|
from .storage import atomic_json, json_bytes, key
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AsyncTestResult:
|
|
package: Path
|
|
request_ref: str
|
|
feedback_ref: str | None = None
|
|
reused: bool = False
|
|
|
|
|
|
class AsyncTestCoordinator:
|
|
"""Trusted controller API, never a public HTTP endpoint or direct LLM tool.
|
|
|
|
A future authenticated web adapter calls confirm_feedback after approval.
|
|
No messages are sent and no clients/bots/labs are controlled here.
|
|
"""
|
|
|
|
def __init__(self, store, *, max_json_bytes=1024 * 1024):
|
|
if type(max_json_bytes) is not int or max_json_bytes <= 0:
|
|
raise LifecycleError("ASYNC_TEST_LIMIT", "Expected a positive JSON byte limit")
|
|
self.store = store
|
|
self.max_json_bytes = max_json_bytes
|
|
|
|
@observed("async_test_submit", identity=False)
|
|
def submit(self, request):
|
|
return self._execute(request, None)
|
|
|
|
@observed("async_test_confirm", identity=False)
|
|
def confirm_feedback(self, feedback):
|
|
return self._execute(None, feedback)
|
|
|
|
def _execute(self, request, feedback):
|
|
document = request if request is not None else feedback
|
|
try:
|
|
validate(document, "async-test-request" if request is not None else "async-test-feedback")
|
|
if len(json_bytes(document)) > self.max_json_bytes:
|
|
raise LifecycleError("ASYNC_TEST_LIMIT", "Test document exceeds operational byte limit")
|
|
case_id, package_id = document["case_id"], document["package_id"]
|
|
with FileLock(self.store.workspace.runtime / "locks" / f"async-tests-{key(case_id)}.lock"):
|
|
snapshot = BudgetManager(self.store).snapshot(case_id, package_id)
|
|
with self.store._locks(case_id):
|
|
latest = self.store.latest(case_id, package_id)
|
|
if latest != snapshot.package:
|
|
raise LifecycleError("GENERATION_CHANGED", "Case changed during test preparation")
|
|
with PackageReader(latest, self.store.limits) as reader:
|
|
return self._submit(reader, request, snapshot) if request is not None else self._confirm(reader, feedback, snapshot)
|
|
except LockBusy:
|
|
raise
|
|
except (CaseBuildError, BudgetError, PackageReadError) as error:
|
|
raise LifecycleError("ASYNC_TEST_INVALID", getattr(error, "code", "PACKAGE_ERROR")) from None
|
|
except ARCHIVE_ERRORS + (ValueError, TypeError, KeyError, RecursionError):
|
|
raise LifecycleError("ASYNC_TEST_INVALID", "Invalid test document or package") from None
|
|
|
|
def _refs(self, document):
|
|
base = f"analysis/async-tests/{document['analysis_revision']:04d}/{key(document['request_id'])}"
|
|
return base + "/request.json", base + "/feedback.json"
|
|
|
|
def _identity(self, reader, document):
|
|
if any(document[field] != reader.manifest[field] for field in ("case_id", "package_id", "analysis_revision")):
|
|
raise LifecycleError("ASYNC_TEST_IDENTITY", "Test document does not belong to the current revision")
|
|
|
|
def _replay(self, reader, ref, document, operation):
|
|
if ref not in reader.files:
|
|
return False
|
|
saved = read_json(reader, ref, "async-test-request" if operation == "async-test-wait" else "async-test-feedback", self.max_json_bytes)
|
|
chain = checkpoint_chain(reader)
|
|
field = "request_ref" if operation == "async-test-wait" else "feedback_ref"
|
|
if saved != document or not any(cp["details"].get("operation") == operation and cp["details"].get(field) == ref
|
|
and cp["details"].get(field.replace("_ref", "_sha256")) == hashlib.sha256(json_bytes(saved)).hexdigest()
|
|
for _, cp in chain):
|
|
raise LifecycleError("ASYNC_TEST_CONFLICT", "Stored test document differs or has no confirming checkpoint")
|
|
self.store._index(reader.path, reader.manifest)
|
|
atomic_json(self.store.workspace.runtime / "current_job.json", None)
|
|
return True
|
|
|
|
def _submit(self, reader, request, snapshot):
|
|
self._identity(reader, request)
|
|
ref, feedback_ref = self._refs(request)
|
|
if self._replay(reader, ref, request, "async-test-wait"):
|
|
return AsyncTestResult(reader.path, ref, feedback_ref if feedback_ref in reader.files else None, True)
|
|
if reader.manifest["state"] != State.INVESTIGATING_HYPOTHESES:
|
|
raise LifecycleError("ASYNC_TEST_PHASE", "Test handoff requires the investigation phase")
|
|
account = snapshot.state["accounts"].get(request["hypothesis_id"], {})
|
|
action = account.get("actions", {}).get(request["action_id"])
|
|
expected_action = "testing.request_human" if request["executor"] == "human" else "testing.request_automated"
|
|
if not action or action["status"] != "reserved" or action["action"] != expected_action:
|
|
raise LifecycleError("ASYNC_TEST_RESERVATION", "A matching pending budget reservation is required")
|
|
if request["budget_ledger_ref"] != snapshot.event_ref:
|
|
raise LifecycleError("ASYNC_TEST_RESERVATION", "Test request must bind the current budget ledger")
|
|
ledger = read_json(reader, snapshot.event_ref, "budget-event", self.max_json_bytes)
|
|
if expected_action not in ledger["policy"]["action_costs"]:
|
|
raise LifecycleError("ASYNC_TEST_COST", "Async tests require an explicitly configured action cost")
|
|
encoded = json_bytes(request)
|
|
details = {"operation": "async-test-wait", "request_ref": ref,
|
|
"request_sha256": hashlib.sha256(encoded).hexdigest(), "automatic_retry": False}
|
|
routing = {"destination": "inbox", "reason": "AWAITING_TEST_FEEDBACK", "retry_not_before": request["deadline"],
|
|
"retry_count": reader.manifest["routing"]["retry_count"] + 1}
|
|
path = self.store._commit(reader, f"async-test:v1:{request['analysis_revision']}:{key(request['request_id'])}:wait",
|
|
hashlib.sha256(encoded).hexdigest(), State.BLOCKED,
|
|
details=details, additions={ref: encoded}, resume_state=State.INVESTIGATING_HYPOTHESES,
|
|
updates={"routing": routing, "artifacts": {"async_test_request": ref}})
|
|
return AsyncTestResult(path, ref)
|
|
|
|
def _confirm(self, reader, feedback, snapshot):
|
|
self._identity(reader, feedback)
|
|
request_ref, ref = self._refs(feedback)
|
|
if self._replay(reader, ref, feedback, "async-test-feedback-confirmed"):
|
|
return AsyncTestResult(reader.path, request_ref, ref, True)
|
|
request = read_json(reader, request_ref, "async-test-request", self.max_json_bytes)
|
|
if any(feedback[field] != request[field] for field in ("request_id", "hypothesis_id", "action_id", "executor")):
|
|
raise LifecycleError("ASYNC_TEST_IDENTITY", "Feedback does not match its test request")
|
|
if (reader.manifest["state"] != State.BLOCKED
|
|
or reader.manifest["routing"]["reason"] != "AWAITING_TEST_FEEDBACK"
|
|
or reader.manifest["artifacts"].get("async_test_request") != request_ref):
|
|
raise LifecycleError("ASYNC_TEST_NOT_WAITING", "Case is not waiting for this test")
|
|
if not self._replay(reader, request_ref, request, "async-test-wait"):
|
|
raise LifecycleError("ASYNC_TEST_CONFLICT", "Missing original test handoff")
|
|
action = snapshot.state["accounts"][request["hypothesis_id"]]["actions"][request["action_id"]]
|
|
if action["status"] != "reserved" or request["budget_ledger_ref"] != snapshot.event_ref:
|
|
raise LifecycleError("ASYNC_TEST_RESERVATION", "Waiting reservation or budget changed")
|
|
for attachment in feedback["attachments"]:
|
|
require_file(reader, attachment["source_ref"])
|
|
with reader._archive.open(attachment["source_ref"]) as stream:
|
|
digest = hashlib.sha256()
|
|
while chunk := stream.read(65536):
|
|
digest.update(chunk)
|
|
if digest.hexdigest() != attachment["sha256"]:
|
|
raise LifecycleError("ASYNC_TEST_ATTACHMENT", "Feedback attachment hash mismatch")
|
|
encoded = json_bytes(feedback)
|
|
# Event-driven resume deliberately bypasses retry_not_before. This is
|
|
# restricted to the matched, confirmed feedback path; ordinary advance
|
|
# cannot resume a test wait, even when its deadline has elapsed.
|
|
path = self.store._commit(reader, f"async-test:v1:{feedback['analysis_revision']}:{key(feedback['request_id'])}:feedback",
|
|
hashlib.sha256(encoded).hexdigest(), State.INVESTIGATING_HYPOTHESES,
|
|
details={"operation": "async-test-feedback-confirmed", "request_ref": request_ref,
|
|
"feedback_ref": ref, "feedback_sha256": hashlib.sha256(encoded).hexdigest()},
|
|
additions={ref: encoded}, updates={"routing": {"destination": "cases", "reason": None,
|
|
"retry_not_before": None}, "artifacts": {"async_test_feedback": ref}})
|
|
return AsyncTestResult(path, request_ref, ref)
|