diff --git a/README.md b/README.md index 1de061f..4fb8092 100644 --- a/README.md +++ b/README.md @@ -35,3 +35,7 @@ Initial hypotheses, revision-wide creation limits and equal budget allocations: Durable investigation budget accounting, configurable costs/rewards and stopping rules: [budget-manager-v1.md](docs/budget-manager-v1.md), with [budget configuration](config/budget.example.toml). + +Case-local asynchronous test requests, confirmed feedback and budget-preserving +event-driven resume: [async-tests-v1.md](docs/async-tests-v1.md). Web UI and external +bot/client adapters are future integrations. diff --git a/config/budget.example.toml b/config/budget.example.toml index 45b3167..2a50729 100644 --- a/config/budget.example.toml +++ b/config/budget.example.toml @@ -13,3 +13,5 @@ max_low_information_steps = 3 # Consecutive settled low-gain actions "source.read_context" = 5 "database.lookup" = 10 "research.lookup" = 20 +"testing.request_human" = 40 +"testing.request_automated" = 20 diff --git a/docs/async-tests-v1.md b/docs/async-tests-v1.md new file mode 100644 index 0000000..e0e86f3 --- /dev/null +++ b/docs/async-tests-v1.md @@ -0,0 +1,106 @@ +# Asynchronous test handoff v1 + +This addition prepares the existing project for externally performed human and +automated tests. No web UI, HTTP feedback endpoint, notifications, bot steering, +client automation, Tool Router or Investigation Loop is implemented here. +`docs/architecture.md` remains unchanged and authoritative: VianoBugAnalysis +does not steer labs. It requests observations through a future external adapter. + +## Durable contract + +`AsyncTestCoordinator` is imported from `viano_bug_analysis.lifecycle.async_tests`. +It is a trusted controller service, not a callable operation exposed directly to +an LLM. The future router must validate requests before calling it. The future +web adapter must authenticate/authorize users and confirm feedback before calling +`confirm_feedback`; a payload claiming `confirmed = true` is not authentication. +User-supplied observations are untrusted data, not executable instructions. + +Two bundled JSON schemas define the case-local exchange: + +- `async-test-request-v1`: stable request/action/hypothesis IDs, case/package IDs, + revision, executor (`human` or `automated`), question, prerequisites, steps, + requested observations, environment, deadline and reserved budget ledger ref. +- `async-test-feedback-v1`: matching identity/executor, observed environment, + tester ID, outcome (`completed` or `unable_to_test`), observations, attachment + references/hashes, and confirming actor/time. Unsupported operations, unknown + fields and unconfirmed feedback are rejected. + +The controller reserves budget first, using `testing.request_human` or +`testing.request_automated`, then calls `submit(request)`. These actions require +explicit cost overrides in the frozen budget policy. The example budget config +sets human tests to 40 units and automated tests to 20. These are tunable initial +values; they do not start any test execution. Existing revisions need their +existing policy and cannot silently adopt a new cost table. + +One outstanding case-level test is allowed at present. A reservation and its +submission are separate durable steps. A failed submission does not silently +refund the reservation: the controller can retry the same valid request or +explicitly fail an undispatched reservation under existing budget rules. Future +delivery/claim/cancellation policies will need explicit contracts; none are +inferred from a deadline or idle time. + +## Waiting and confirmation + +```text +INVESTIGATING_HYPOTHESES + matching reserved action + -> request stored + BLOCKED/AWAITING_TEST_FEEDBACK + -> worker continues with other cases + -> authenticated external adapter confirms matching feedback + -> feedback stored + INVESTIGATING_HYPOTHESES (same analysis_revision) + -> future controller evaluates result and settles the reserved action +``` + +Submission stores an immutable request under +`analysis/async-tests/NNNN//request.json`, sets the manifest +pointer `async_test_request`, and publishes a wait checkpoint. The checkpoint +has `automatic_retry = false`. The deadline is recorded as `retry_not_before` +for scheduling visibility, but it does not authorize an automatic resume. +The inbox worker explicitly skips `AWAITING_TEST_FEEDBACK`; ordinary lifecycle +`advance` cannot resume such a case. Deadline expiry neither consumes budget +nor creates a failure, new analysis revision, refund or replacement request. +Deadline escalation/notifications are future adapter/controller policy. + +`confirm_feedback(feedback)` verifies the original handoff, matching identities, +pending reservation, unchanged ledger, schemas and attachment hashes. It stores +`feedback.json`, sets `async_test_feedback`, and resumes the investigation even +before `retry_not_before`. It deliberately uses a narrowly controlled event-driven +resume path rather than disabling retry protection for other blocked reasons. + +While waiting, remaining balance, expenses, rewards, low-information counters +and pending reservations remain unchanged. Confirmation also leaves the budget +ledger unchanged. Only subsequent `record_result`/`settle` applies novelty, +semantic scoring and possible reward. Waiting hours never cost extra credits; +confirmation never restores start budgets. Unavailable tests can return a +confirmed `unable_to_test` observation and be assessed as a low-information +result. Unrelated post-diagnosis review or patch rejection still uses the +existing `REOPENED` path and creates a new revision. + +Both APIs are idempotent. The same document reuses the current package; +conflicting content under the same request ID is rejected. Unknown, wrong-case, +wrong-revision, unmatched and unconfirmed feedback cannot wake the case. Events +increase only `package_generation` and preserve all previous payloads. A case +waiting for external feedback can be admitted on another service before its +deadline and still stays blocked until confirmed feedback arrives. + +## Future web/bot adapter responsibilities + +The UI will list persisted requests, allow authorized testers to claim them, +capture observations and upload attachments, and call the confirmation API. +External adapters need reliable/idempotent delivery, authentication, authorization, +attachment size/type policy and notification/escalation handling. Delivery state +must not replace the case-local original request and feedback history. Draft +feedback stays outside the authoritative package until confirmed; no console +entry or SSH workflow is required. + +Attachments must already exist in the package when confirmation is called, with +matching hashes. The future upload importer will append them using the existing +safe writer, preferably under the same request's `attachments/` directory. +Unknown files are preserved. Checkpoint-declared request/feedback artifacts and +their async-test attachment references are excluded from raw CaseBuilder input +identity, so expected investigation feedback does not force case regeneration. +Existing original-source attachments retain their original input classification. + +The coordinator's configurable `max_json_bytes` (default 1 MiB) is an operational +safeguard, not a permanent limit of either v1 format. Case locking, generation +comparison and atomic package publication use the existing lifecycle foundation. +No lock is held while a human, bot, network provider or model is working. diff --git a/docs/budget-manager-v1.md b/docs/budget-manager-v1.md index 00861c7..aa30bc3 100644 --- a/docs/budget-manager-v1.md +++ b/docs/budget-manager-v1.md @@ -190,3 +190,9 @@ incremental replay optimizations are a later performance task. The CaseAssembler excludes checkpoint-declared budget events from raw case input identity, preserving replay of earlier phases. Undeclared/unknown files remain preserved and part of raw input inventory as before. + +For asynchronous external testing, see [async-tests-v1.md](async-tests-v1.md). +`BLOCKED` pauses the case, not accounting history: balances and pending +reservations survive unchanged. Confirmed feedback resumes the same revision; +only subsequent result assessment changes its budget. Human/automated tests +have separately configurable explicit costs, with no idle-time charges. diff --git a/src/viano_bug_analysis/case/assembler.py b/src/viano_bug_analysis/case/assembler.py index 5a2fcd1..1e7e4a7 100644 --- a/src/viano_bug_analysis/case/assembler.py +++ b/src/viano_bug_analysis/case/assembler.py @@ -84,6 +84,22 @@ class CaseAssembler: prior_refs = [] for _, checkpoint in reversed(chain): details = checkpoint["details"] + if details.get("operation") in {"async-test-wait", "async-test-feedback-confirmed"}: + for field in ("request_ref", "feedback_ref"): + ref = details.get(field) + if ref is None: + continue + if not isinstance(ref, str) or not ref.startswith("analysis/async-tests/"): + raise CaseBuildError("PREVIOUS_CASE", "Invalid async test checkpoint") + require_file(reader, ref) + excluded.add(ref) + if field == "feedback_ref": + feedback = load(ref, "async-test-feedback") + for attachment in feedback["attachments"]: + path = attachment["source_ref"] + require_file(reader, path) + if path.startswith("analysis/async-tests/"): + excluded.add(path) if details.get("operation") == "budget-event": ref = details.get("event_ref") if not isinstance(ref, str) or not ref.startswith("analysis/budgets/"): diff --git a/src/viano_bug_analysis/lifecycle/async_tests.py b/src/viano_bug_analysis/lifecycle/async_tests.py new file mode 100644 index 0000000..4a63910 --- /dev/null +++ b/src/viano_bug_analysis/lifecycle/async_tests.py @@ -0,0 +1,154 @@ +"""Case-local async test handoff; external delivery and authentication are adapters.""" + +import hashlib +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 + + def submit(self, request): + return self._execute(request, None) + + 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) diff --git a/src/viano_bug_analysis/lifecycle/checkpoints.py b/src/viano_bug_analysis/lifecycle/checkpoints.py index 91fee92..b390ea2 100644 --- a/src/viano_bug_analysis/lifecycle/checkpoints.py +++ b/src/viano_bug_analysis/lifecycle/checkpoints.py @@ -209,7 +209,7 @@ class CheckpointStore: raise LifecycleError("CHECKPOINT_INVALID", "Blocked input requires its resume checkpoint") if not manifest["routing"]["reason"] or manifest["routing"]["retry_not_before"] is None: raise LifecycleError("INVALID_RETRY", "Blocked input requires a reason and retry timestamp") - if not self.retry_due(manifest): + if not self.retry_due(manifest) and manifest["routing"]["reason"] != "AWAITING_TEST_FEEDBACK": return "DEFERRED", source state = State.VALIDATED if manifest["state"] == State.NEW else manifest["state"] step_id = f"intake:{int(manifest['package_generation'])}:{digest}" @@ -267,6 +267,8 @@ class CheckpointStore: if expected_generation is not None and manifest["package_generation"] != expected_generation: raise LifecycleError("GENERATION_CHANGED", "Case changed while artifacts were being prepared") resume = chain[0][1]["resume_state"] if manifest["state"] == State.BLOCKED else None + if manifest["state"] == State.BLOCKED and manifest["routing"]["reason"] == "AWAITING_TEST_FEEDBACK": + raise LifecycleError("EXTERNAL_FEEDBACK_REQUIRED", "Confirmed test feedback is required to resume this case") check_transition(manifest["state"], target, resume) if manifest["state"] == State.BLOCKED and not self.retry_due(manifest): raise LifecycleError("RETRY_NOT_DUE", "Retry timestamp has not been reached") diff --git a/src/viano_bug_analysis/lifecycle/queue.py b/src/viano_bug_analysis/lifecycle/queue.py index 42d3ab9..ade8afb 100644 --- a/src/viano_bug_analysis/lifecycle/queue.py +++ b/src/viano_bug_analysis/lifecycle/queue.py @@ -146,7 +146,8 @@ class InboxProcessor: path = self.store.latest(case_id, package_id) with PackageReader(path, self.store.limits) as reader: manifest = reader.manifest - if manifest["state"] != State.BLOCKED or not self.store.retry_due(manifest): + if (manifest["state"] != State.BLOCKED or not self.store.retry_due(manifest) + or manifest["routing"]["reason"] == "AWAITING_TEST_FEEDBACK"): continue checkpoint = checkpoint_chain(reader)[0][1] if checkpoint["details"].get("automatic_retry") is False: diff --git a/src/viano_bug_analysis/schemas/async-test-feedback-v1.schema.json b/src/viano_bug_analysis/schemas/async-test-feedback-v1.schema.json new file mode 100644 index 0000000..fd22f5b --- /dev/null +++ b/src/viano_bug_analysis/schemas/async-test-feedback-v1.schema.json @@ -0,0 +1,134 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "type": "object", + "additionalProperties": false, + "required": [ + "format_version", + "case_id", + "package_id", + "analysis_revision", + "request_id", + "hypothesis_id", + "action_id", + "executor", + "format", + "outcome", + "observations", + "environment", + "tester_id", + "confirmation", + "attachments" + ], + "properties": { + "format_version": { + "const": 1 + }, + "case_id": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + }, + "package_id": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + }, + "analysis_revision": { + "type": "integer", + "minimum": 1 + }, + "request_id": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + }, + "hypothesis_id": { + "type": "string", + "pattern": "^H[1-9][0-9]*$" + }, + "action_id": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + }, + "executor": { + "enum": [ + "human", + "automated" + ] + }, + "format": { + "const": "viano-async-test-feedback" + }, + "outcome": { + "enum": [ + "completed", + "unable_to_test" + ] + }, + "observations": { + "type": "array", + "minItems": 1, + "items": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + } + }, + "environment": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + }, + "tester_id": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + }, + "confirmation": { + "type": "object", + "additionalProperties": false, + "required": [ + "confirmed", + "confirmed_by", + "confirmed_at" + ], + "properties": { + "confirmed": { + "const": true + }, + "confirmed_by": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + }, + "confirmed_at": { + "type": "string", + "format": "date-time" + } + } + }, + "attachments": { + "type": "array", + "items": { + "type": "object", + "additionalProperties": false, + "required": [ + "source_ref", + "sha256" + ], + "properties": { + "source_ref": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + }, + "sha256": { + "type": "string", + "pattern": "^[0-9a-f]{64}$" + } + } + } + } + } +} diff --git a/src/viano_bug_analysis/schemas/async-test-request-v1.schema.json b/src/viano_bug_analysis/schemas/async-test-request-v1.schema.json new file mode 100644 index 0000000..83b4392 --- /dev/null +++ b/src/viano_bug_analysis/schemas/async-test-request-v1.schema.json @@ -0,0 +1,111 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "type": "object", + "additionalProperties": false, + "required": [ + "format_version", + "case_id", + "package_id", + "analysis_revision", + "request_id", + "hypothesis_id", + "action_id", + "executor", + "format", + "question", + "steps", + "prerequisites", + "requested_observations", + "environment", + "deadline", + "budget_ledger_ref" + ], + "properties": { + "format_version": { + "const": 1 + }, + "case_id": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + }, + "package_id": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + }, + "analysis_revision": { + "type": "integer", + "minimum": 1 + }, + "request_id": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + }, + "hypothesis_id": { + "type": "string", + "pattern": "^H[1-9][0-9]*$" + }, + "action_id": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + }, + "executor": { + "enum": [ + "human", + "automated" + ] + }, + "format": { + "const": "viano-async-test-request" + }, + "question": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + }, + "steps": { + "type": "array", + "minItems": 1, + "items": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + } + }, + "prerequisites": { + "type": "array", + "minItems": 1, + "items": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + } + }, + "requested_observations": { + "type": "array", + "minItems": 1, + "items": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + } + }, + "environment": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + }, + "deadline": { + "type": "string", + "format": "date-time" + }, + "budget_ledger_ref": { + "type": "string", + "minLength": 1, + "pattern": "\\S" + } + } +} diff --git a/tests/test_async_tests.py b/tests/test_async_tests.py new file mode 100644 index 0000000..1022839 --- /dev/null +++ b/tests/test_async_tests.py @@ -0,0 +1,117 @@ +import copy +import json + +import pytest + +from viano_bug_analysis.budget import BudgetConfig, BudgetManager +from viano_bug_analysis.lifecycle import CheckpointStore, InboxProcessor, LifecycleError, State, Workspace +from viano_bug_analysis.lifecycle.async_tests import AsyncTestCoordinator +from viano_bug_analysis.package import PackageReader, PackageValidator +from test_budget_manager import hypotheses +from test_case_builder import case_setup +from test_case_interpreter import built_case +from test_hypothesis_builder import interpreted_case + +CASE = ("case-A", "stable-package") + + +def setup(hypotheses, executor="human", deadline="2099-01-01T00:00:00Z"): + store = hypotheses[0] + budget = BudgetManager(store, BudgetConfig(action_costs={"testing.request_human":40,"testing.request_automated":20})) + budget.initialize(*CASE) + action = "testing.request_human" if executor == "human" else "testing.request_automated" + budget.reserve(*CASE, "H1", "test-action", action) + store.advance(*CASE, State.INVESTIGATING_HYPOTHESES, step_id="begin-investigation") + before = budget.snapshot(*CASE) + request = {"format":"viano-async-test-request","format_version":1,"case_id":CASE[0],"package_id":CASE[1], + "analysis_revision":1,"request_id":"test-1","hypothesis_id":"H1","action_id":"test-action", + "executor":executor,"question":"What happens after interacting?","steps":["Interact once"], + "prerequisites":["Use the specified test environment"],"requested_observations":["Record the observed transition"], + "environment":"test-build-A","deadline":deadline,"budget_ledger_ref":before.event_ref} + feedback = {name:request[name] for name in ("format_version","case_id","package_id","analysis_revision","request_id","hypothesis_id","action_id","executor")} + feedback.update(format="viano-async-test-feedback",outcome="completed",observations=["The transition was absent"], + environment="test-build-A",tester_id="tester-1",attachments=[], + confirmation={"confirmed":True,"confirmed_by":"reviewer-1","confirmed_at":"2026-10-14T12:00:00Z"}) + return store,budget,AsyncTestCoordinator(store),before,request,feedback + + +@pytest.mark.parametrize("executor", ["human","automated"]) +def test_wait_confirmation_preserves_budget_revision_and_payloads(hypotheses, executor): + store,budget,coordinator,before,request,feedback = setup(hypotheses,executor) + blocked = coordinator.submit(request) + assert coordinator.submit(request).reused + assert budget.snapshot(*CASE).state == before.state + with PackageReader(blocked.package) as reader: + assert reader.manifest["state"] == State.BLOCKED + assert reader.manifest["routing"]["reason"] == "AWAITING_TEST_FEEDBACK" + with pytest.raises(LifecycleError, match="Confirmed test feedback"): + store.advance(*CASE,State.INVESTIGATING_HYPOTHESES,step_id="timer-resume") + resumed = coordinator.confirm_feedback(feedback) + assert coordinator.confirm_feedback(feedback).reused + assert budget.snapshot(*CASE).state == before.state + assert PackageValidator().validate(resumed.package).valid + with PackageReader(before.package) as old, PackageReader(resumed.package) as new: + assert new.manifest["state"] == State.INVESTIGATING_HYPOTHESES + assert new.manifest["analysis_revision"] == 1 + assert new.manifest["package_generation"] == old.manifest["package_generation"] + 2 + assert new.manifest["routing"]["retry_not_before"] is None + for name in old.files: + if name != "manifest.json": + assert old.read_bytes(name) == new.read_bytes(name) + # Feedback is an investigation result, not a reason to regenerate the case. + assert hypotheses[1].build(*CASE).reused + budget.record_result(*CASE,"H1","test-action",{"observations":feedback["observations"]},source_refs=[resumed.feedback_ref]) + assessed = budget.settle(*CASE,"H1","test-action",semantic_score=50,reason="Useful observation") + cost = 40 if executor == "human" else 20 + assert assessed.state["accounts"]["H1"]["remaining"] == 100-cost+cost*4//10 + + +def test_deadline_does_not_resume_or_charge_idle_case(hypotheses): + store,budget,coordinator,before,request,feedback = setup(hypotheses,deadline="2000-01-01T00:00:00Z") + coordinator.submit(request) + outcomes = InboxProcessor(store.workspace).poll() + assert not any(outcome.status == "RESUMED" for outcome in outcomes) + assert budget.snapshot(*CASE).state == before.state + coordinator.confirm_feedback(feedback) + assert budget.snapshot(*CASE).state == before.state + + +@pytest.mark.parametrize("change", [ + lambda f: f.update(analysis_revision=2), + lambda f: f.update(hypothesis_id="H2"), + lambda f: f.update(request_id="other"), + lambda f: f["confirmation"].update(confirmed=False), + lambda f: f.update(attachments=[{"source_ref":"../escape","sha256":"0"*64}]), +]) +def test_invalid_feedback_never_wakes_or_changes_budget(hypotheses, change): + store,budget,coordinator,before,request,feedback = setup(hypotheses) + blocked = coordinator.submit(request) + change(feedback) + with pytest.raises(LifecycleError): + coordinator.confirm_feedback(feedback) + assert budget.snapshot(*CASE).package == blocked.package + assert budget.snapshot(*CASE).state == before.state + + +def test_portable_wait_feedback_and_conflicting_replay(hypotheses,tmp_path): + store,budget,coordinator,before,request,feedback = setup(hypotheses) + blocked = coordinator.submit(request) + portable = CheckpointStore(Workspace(tmp_path / "other-service")) + status, managed = portable.intake(blocked.package) + assert status == "ACCEPTED" + assert BudgetManager(portable).snapshot(*CASE).state == before.state + remote_resumed = AsyncTestCoordinator(portable).confirm_feedback(feedback) + with PackageReader(remote_resumed.package) as reader: + assert reader.manifest["state"] == State.INVESTIGATING_HYPOTHESES + with PackageReader(blocked.package) as reader: + assert reader.manifest["artifacts"]["async_test_request"] == blocked.request_ref + changed = copy.deepcopy(request) + changed["question"] = "A different test" + with pytest.raises(LifecycleError): + coordinator.submit(changed) + resumed = coordinator.confirm_feedback(feedback) + changed = copy.deepcopy(feedback) + changed["observations"] = ["Different feedback"] + with pytest.raises(LifecycleError): + coordinator.confirm_feedback(changed) + assert budget.snapshot(*CASE).package == resumed.package