Prepare asynchronous test handoffs with budget-preserving resume

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.

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

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/<hashed-request-id>/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.

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.

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/"):

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)

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")

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:

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}$"
+          }
+        }
+      }
+    }
+  }
+}

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"
+    }
+  }
+}

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