"""Protected, single-task work intents inside an existing AutonomousAgent custody. No store, credential lifecycle, scheduler, execution or payment authority is added. Explicit resolution checks the original operation or fences it against future execution. Once resolution begins, restarts resume only that resolution request, never the original mutation. Resolution consumes its own cleanup-pool attempt; prior unknown charges remain held even when the logical outcome is resolved. The adapter's send_once MUST perform exactly one configured CommonsClient request and validate its authenticated response before returning a body-free reference. There is no /me, capabilities fetch, wait or retry loop hidden in this helper. One immutable task budget supports finite sequential mutation intents. Atomic advancement closes only a logically terminal task and explicitly selects its successor. Retired spent/unknown totals survive in fixed-size sealed accounting; the keyed chain commits to discarded detail but cannot reconstruct that detail. Compacted unknown costs cannot later be refunded or reconciled per attempt. Generation-qualified tasks are distinct even when caller labels repeat. These local task ceilings are not an identity-lifetime account cap or agreement. Only the one pending intent retains its body/key, in the existing sealed envelope. HMAC bindings are private and use a random key generated inside that same state. Explicit GETs share that task's finite ledger without occupying mutation pending. Their sealed records contain only keyed request bindings and body-free checkpoints; selected public observations are returned transiently after the completion save. Runtime accounting reserves and charges 20,000 milliseconds per physical attempt, without measured refunds. Standard Windows work transport has an 18-second abort deadline plus 2-second teardown allocation, with retry gated on confirmed exit. Explicitly injected client factories own their transport. OS stalls prevent an unconditional wall-clock guarantee. Observed overruns halt without releasing exposure. Unknown earlier attempts remain unknown even if a later exact-intent replay works. The latter proves the logical result, not which earlier transport attempt ran. Durability depends on successful existing two-copy custody saves. That custody cannot independently detect simultaneous rollback of both valid sealed copies. No production/native-host integration or remote resource enforcement is claimed. Work JSON is capped at 64 KiB (custody's ASCII-escaped JSON encoding). Every work save leaves 32 KiB of the 128 KiB envelope unused for identity lifecycle expansion; dispatch also reserves completion headroom within both limits. This is a bounded allocation, NOT a proven maximum for arbitrary profile/principal sizes, nested uncertain recovery chains, or unrelated state. Lifecycle methods still enforce the canonical envelope limit. Worst-case chain sizing remains an integration gate. NO JUDGMENT WHATSOEVER. No participant evaluation, ranking or memory promotion. """ from __future__ import annotations import copy from datetime import datetime, timezone from functools import wraps import hashlib import hmac import json import re import secrets import time import concresca_work_budget as budget from concresca_autonomous_agent import ( AgentError, AutonomousCommonsClient, _json, _MAX_BYTES, _WORK_TEARDOWN_MS, _redacted as _agent_redacted, ) from concresca_commons_client import CommonsClientError, _idempotency, _segment SCHEMA = "concresca.work_controller.v1" STATE_KEY = "workController" ATTEMPT_RUNTIME_MS = 20_000 MAX_REFERENCE_BYTES = 4096 MAX_WORK_BYTES = 64 * 1024 LIFECYCLE_RESERVE_BYTES = 32 * 1024 COMPLETION_RESERVE_BYTES = MAX_REFERENCE_BYTES + 2048 MAX_READ_CHECKPOINT_BYTES = 8192 MAX_READ_OBSERVATION_BYTES = 1024 * 1024 RESOLUTION_PATH = "/api/matm/commons/work-resolution" RESOLUTION_SCHEMA = "concresca.work_resolution.v1" _ID = re.compile(r"[A-Za-z0-9][A-Za-z0-9._~-]{0,63}") _SHA = re.compile(r"[0-9a-f]{64}") _WITHDRAW = re.compile(r"/api/matm/commons/work/commonswork-[0-9a-f]{24}/withdrawal") _ACTION = re.compile(r"/api/matm/commons/work-proposals/commonsproposal-[0-9a-f]{24}/actions") _READ_OBJECT = re.compile(r"/api/matm/commons/(work/commonswork-[0-9a-f]{24}|work-proposals/commonsproposal-[0-9a-f]{24})") _READ_LIST = re.compile(r"/api/matm/commons/work(?:/commonswork-[0-9a-f]{24}/proposals)?") _BINDING = {"baseUrl", "agentId", "workspaceId", "projectId"} _ROOT = {"schemaVersion", "taskId", "binding", "secret", "ledger", "intents", "pending", "continuityCredentialId", "halted", "mac"} _INTENT = {"intentId", "requestDigest", "bindingDigest", "keyScopeDigest", "pool", "attemptIds", "state", "resultDigest", "resultReference"} _PENDING = {"intentId", "method", "path", "body", "key", "bindingDigest", "notBeforeMilliseconds"} def _redacted(function): @wraps(function) def safe(*args, **kwargs): try: return function(*args, **kwargs) except budget.BudgetError as exc: # Ledger codes are fixed local constants, not remote/body strings. raise AgentError(exc.code) from None return _agent_redacted(safe) def _require(condition, code="work_state_invalid"): if not condition: raise AgentError(code) def _budget_call(function, *args, **kwargs): # Translate before leaving transaction(): its generic exception boundary # otherwise correctly redacts these into an unhelpful custody_unavailable. try: return function(*args, **kwargs) except budget.BudgetError as exc: raise AgentError(exc.code) from None def _fields(value, fields): _require(type(value) is dict and set(value) == fields) def _integer(value, maximum=budget.MAX_INTEGER): _require(type(value) is int and 0 <= value <= maximum) return value def _identifier(value): _require(type(value) is str and _ID.fullmatch(value) is not None) def _sha(value): _require(type(value) is str and _SHA.fullmatch(value) is not None) def _mac(work, domain, value): # Generation zero preserves every legacy binding byte. Lifecycle commitments # use the stable custody secret so an exact advance retry survives rollover. material = [SCHEMA, domain, value] generation = _generation(work) if generation and not domain.startswith("lifecycle-"): material = [SCHEMA, domain, generation, value] return hmac.new(bytes.fromhex(work["secret"]), _json(material), hashlib.sha256).hexdigest() def _generation(work): return work.get("taskLifecycle", {}).get("generation", 0) def _check_generation(work, expected): actual = _generation(work) if expected is None: _require(actual == 0, "work_task_generation_required") else: _integer(expected) _require(expected == actual, "work_task_generation_conflict") def _remote_key(work, kind, value): if not _generation(work): return secrets.token_urlsafe(32) return "concresca-work-g%d-%s" % (_generation(work), _mac(work, "wire-key-" + kind, value)) def _retired_zero(): return {"tasks": 0, "completedIntents": 0, "preventedIntents": 0, "completedReads": 0, "unknownReads": 0, "pools": {pool: {state: {"attempts": 0, "requests": 0, "runtimeMilliseconds": 0} for state in ("spent", "unknown")} for pool in ("work", "cleanup")}} def _add_counter(first, second): result = first + second _require(type(result) is int and 0 <= result <= budget.MAX_INTEGER, "work_retirement_counter_overflow") return result def _scope(work): return _mac(work, "task", [work["taskId"], work["binding"]]) def _intent_binding(work, identifier, descriptor, key): return _mac(work, "intent", [work["taskId"], work["binding"], identifier, descriptor["method"], descriptor["path"], descriptor["body"], key]) def _pool(descriptor): if _WITHDRAW.fullmatch(descriptor["path"]): return "cleanup" if (_ACTION.fullmatch(descriptor["path"]) and descriptor["body"].get("action") in {"withdraw_terms", "withdraw_delivery", "withdraw_decision"}): return "cleanup" return "work" class WorkController: """Thin integration helper; instantiate with the existing AutonomousAgent. adapter: module/object exposing validate_request(method, path, body) and send_once(client, descriptor, binding, idempotency_key). Descriptor contains method/path/body/expectedSchema/expectedStatus; additional adapter metadata is allowed but is regenerated, not trusted from custody. send_once returns a bounded, validated body-free result reference (not the wire response). A 401 requires external same-identity credential continuity before resuming. Main's lifecycle integration must preserve STATE_KEY on every state rewrite; this helper never calls ensure_active/resume/_refresh or repairs credentials. """ def __init__(self, agent, *, adapter=None, clock_ms=None, monotonic_ns=None): self._agent = agent self._adapter_value = adapter self._clock = clock_ms or (lambda: time.time_ns() // 1_000_000) self._monotonic = monotonic_ns or time.monotonic_ns def __repr__(self): return "WorkController(values_redacted=True)" def _adapter(self): if self._adapter_value is None: import concresca_work_client return concresca_work_client return self._adapter_value def _descriptor(self, method, path, body): _require(method == "POST" and type(method) is str, "work_mutation_required") _require(path != RESOLUTION_PATH, "work_resolution_requires_pending_intent") # Reject Python-only JSON shapes before a helper can silently normalize. self._json_types(body) descriptor = self._adapter().validate_request(method, path, copy.deepcopy(body)) _require(type(descriptor) is dict and {"method", "path", "body", "expectedSchema", "expectedStatus"} <= set(descriptor), "work_adapter_contract_invalid") _require(descriptor["method"] == method and descriptor["path"] == path and _json(descriptor["body"]) == _json(body), "work_adapter_contract_invalid") _require(type(descriptor["expectedStatus"]) is int and descriptor["expectedStatus"] in (200, 201) and type(descriptor["expectedSchema"]) is str, "work_adapter_contract_invalid") _require(type(body) is dict and type(path) is str, "work_adapter_contract_invalid") return copy.deepcopy(descriptor) def _resolution_descriptor(self, pending): original = self._descriptor(pending["method"], pending["path"], pending["body"]) result = self._adapter().prepare_resolution(original, pending["key"]) expected_body = {"schemaVersion": RESOLUTION_SCHEMA, "originalPath": pending["path"], "originalBody": pending["body"], "originalIdempotencyKey": pending["key"]} _require(type(result) is dict and result.get("method") == "POST" and result.get("path") == RESOLUTION_PATH and result.get("expectedSchema") == RESOLUTION_SCHEMA and type(result.get("expectedStatus")) is int and result["expectedStatus"] == 200 and _json(result.get("body")) == _json(expected_body), "work_adapter_contract_invalid") return copy.deepcopy(result) def _read_descriptor(self, path): _require(type(path) is str and len(path) <= 2300, "work_read_path_invalid") bare = path.split("?", 1)[0] _require(bool(_READ_LIST.fullmatch(bare)) or ("?" not in path and bool(_READ_OBJECT.fullmatch(bare))), "work_read_path_invalid") result = self._adapter().validate_request("GET", path, None) _require(type(result) is dict and result.get("method") == "GET" and result.get("path") == path and "body" in result and result["body"] is None and type(result.get("expectedStatus")) is int and result["expectedStatus"] == 200 and type(result.get("expectedSchema")) is str, "work_adapter_contract_invalid") return copy.deepcopy(result) @staticmethod def _read_pool(work, path, cleanup_for): if cleanup_for is None: return "work" _identifier(cleanup_for) _require(_READ_OBJECT.fullmatch(path) is not None, "work_cleanup_read_scope_invalid") intent = next((i for i in work["intents"] if i["intentId"] == cleanup_for), None) _require(intent is not None, "work_cleanup_read_scope_invalid") pending = work["pending"] if pending is not None and pending["intentId"] == cleanup_for: # A proposal-create parent is NOT the resource of that mutation. match = re.fullmatch(r"(/api/matm/commons/work/commonswork-[0-9a-f]{24})/(corrections|withdrawal)", pending["path"]) match = match or re.fullmatch(r"(/api/matm/commons/work-proposals/commonsproposal-[0-9a-f]{24})/actions", pending["path"]) _require(match is not None and match[1] == path, "work_cleanup_read_scope_invalid") else: _require(intent["state"] == "completed", "work_cleanup_read_scope_invalid") reference = intent["resultReference"] if reference.get("resourceKind") == "work_resolution": reference = reference.get("originalResult") _require(type(reference) is dict, "work_cleanup_read_scope_invalid") prefix = {"work_opportunity": "/api/matm/commons/work/", "work_proposal": "/api/matm/commons/work-proposals/"}.get(reference.get("resourceKind")) _require(prefix is not None and type(reference.get("resourceId")) is str and prefix + reference["resourceId"] == path, "work_cleanup_read_scope_invalid") return "cleanup" def _read_checkpoint(self, checkpoint, binding): common = {"schemaVersion", "method", "operation", "cleanup", "httpStatus", "action", "workspaceId", "projectId", "actorAgentId", "receiptId", "operationRevision", "idempotentReplay", "resourceKind", "opportunityId", "observationSource", "consumptionAcknowledged"} _require(type(checkpoint) is dict, "work_read_checkpoint_invalid") page = "itemIds" in checkpoint proposal = checkpoint.get("resourceKind") == "work_proposal" extra = ({"itemIds", "count", "hasMore", "nextCursor"} if page else {"revision", "state", "ownState", "termsSha256", "requesterAgentId", "resourceId"} | ({"providerAgentId", "proposedByAgentId", "deliverySha256", "opportunityRevision", "opportunityTermsSha256"} if proposal else {"termsAuthorAgentId", "selectedProposalId"})) _fields(checkpoint, common | extra) self._json_types(checkpoint) _require(len(_json(checkpoint)) <= (MAX_READ_CHECKPOINT_BYTES if page else MAX_REFERENCE_BYTES), "work_read_checkpoint_invalid") _require(checkpoint["schemaVersion"] == "concresca.work_read_checkpoint.v1" and checkpoint["resourceKind"] in {"work_opportunity", "work_proposal"} and checkpoint["method"] == "GET" and checkpoint["httpStatus"] == 200 and type(checkpoint["httpStatus"]) is int and checkpoint["cleanup"] is False and checkpoint["action"] is None and checkpoint["receiptId"] is None and checkpoint["operationRevision"] is None and checkpoint["idempotentReplay"] is False and checkpoint["observationSource"] == "fresh_validated_response" and checkpoint["consumptionAcknowledged"] is False and checkpoint["operation"] == "work-" + ("proposal" if proposal else "opportunity") + ("-list" if page else "-get") and checkpoint["actorAgentId"] == binding["agentId"] and all(checkpoint[k] == binding[k] for k in ("workspaceId", "projectId")), "work_read_checkpoint_invalid") pattern = r"commonsproposal-[0-9a-f]{24}" if proposal else r"commonswork-[0-9a-f]{24}" opportunity = checkpoint["opportunityId"] _require((page and not proposal and opportunity is None) or (type(opportunity) is str and re.fullmatch(r"commonswork-[0-9a-f]{24}", opportunity)), "work_read_checkpoint_invalid") if page: ids = checkpoint["itemIds"] _require(type(ids) is list and len(ids) <= 50 and all(type(i) is str and re.fullmatch(pattern, i) for i in ids) and ids == sorted(set(ids)) and type(checkpoint["count"]) is int and checkpoint["count"] == len(ids) and type(checkpoint["hasMore"]) is bool, "work_read_checkpoint_invalid") cursor = checkpoint["nextCursor"] _require((not checkpoint["hasMore"] and cursor is None) or (checkpoint["hasMore"] and bool(ids) and type(cursor) is str and len(cursor) <= 2048 and re.fullmatch(r"cw1\.[A-Za-z0-9_-]+\.[0-9a-f]{64}", cursor)), "work_read_checkpoint_invalid") else: _require(type(checkpoint["resourceId"]) is str and re.fullmatch(pattern, checkpoint["resourceId"]), "work_read_checkpoint_invalid") _require(1 <= _integer(checkpoint["revision"], 136), "work_read_checkpoint_invalid") _sha(checkpoint["termsSha256"]) states = ({"proposed", "committed", "refused", "cancelled", "submitted", "accepted", "disputed", "withdrawn", "delivery_withdrawn"} if proposal else {"open", "assigned", "withdrawn"}) _require(type(checkpoint["ownState"]) is str and checkpoint["ownState"] in states and type(checkpoint["state"]) is str and checkpoint["state"] in states | {"expired", "retention_expired"}, "work_read_checkpoint_invalid") _segment(checkpoint["requesterAgentId"], "requesterAgentId") if proposal: for field in ("providerAgentId", "proposedByAgentId"): _segment(checkpoint[field], field) _require(checkpoint["providerAgentId"] != checkpoint["requesterAgentId"] and checkpoint["proposedByAgentId"] in {checkpoint["providerAgentId"], checkpoint["requesterAgentId"]}, "work_read_checkpoint_invalid") _require(1 <= _integer(checkpoint["opportunityRevision"], 136), "work_read_checkpoint_invalid") _sha(checkpoint["opportunityTermsSha256"]) if checkpoint["deliverySha256"] is not None: _sha(checkpoint["deliverySha256"]) else: _require(checkpoint["termsAuthorAgentId"] == checkpoint["requesterAgentId"] and checkpoint["opportunityId"] == checkpoint["resourceId"], "work_read_checkpoint_invalid") selected = checkpoint["selectedProposalId"] _require(selected is None or (type(selected) is str and re.fullmatch(r"commonsproposal-[0-9a-f]{24}", selected)), "work_read_checkpoint_invalid") return copy.deepcopy(checkpoint) @staticmethod def _resolution_attempt(work, intent, operation): return ("resolutionBinding" in intent and operation["bindingDigest"] == _mac(work, "resolution-attempt", [intent["resolutionBinding"], operation["operationId"]])) @staticmethod def _json_types(value, depth=0): _require(depth <= 24) if type(value) is dict: _require(len(value) <= 1024 and all(type(k) is str for k in value)) for child in value.values(): WorkController._json_types(child, depth + 1) elif type(value) is list: _require(len(value) <= 1024) for child in value: WorkController._json_types(child, depth + 1) else: _require(value is None or type(value) in (str, int, bool)) if depth == 0: _json(value) def _validate(self, state): self._agent._validate_state(state) return self._validate_work(state.get(STATE_KEY), state["binding"]) def _validate_lifecycle(self, lifecycle): _fields(lifecycle, {"generation", "retired", "chainHead", "lastAdvanceDigest"}) generation = _integer(lifecycle["generation"]) _require(generation > 0) _sha(lifecycle["chainHead"]) _sha(lifecycle["lastAdvanceDigest"]) retired = lifecycle["retired"] _fields(retired, set(_retired_zero())) for field in ("tasks", "completedIntents", "preventedIntents", "completedReads", "unknownReads"): _integer(retired[field]) _require(retired["tasks"] == generation) _fields(retired["pools"], {"work", "cleanup"}) spent, unknown = 0, 0 for pool in ("work", "cleanup"): _fields(retired["pools"][pool], {"spent", "unknown"}) for category in ("spent", "unknown"): values = retired["pools"][pool][category] _fields(values, {"attempts", "requests", "runtimeMilliseconds"}) for value in values.values(): _integer(value) _require(values["requests"] == values["attempts"] and values["runtimeMilliseconds"] == values["attempts"] * ATTEMPT_RUNTIME_MS) spent += retired["pools"][pool]["spent"]["attempts"] unknown += retired["pools"][pool]["unknown"]["attempts"] _require(spent == retired["completedIntents"] + retired["preventedIntents"] + retired["completedReads"] and retired["unknownReads"] <= unknown) _require(len(_json(lifecycle)) <= 4096, "work_component_limit_exceeded") def _validate_work(self, work, binding): _require(type(work) is dict and _ROOT <= set(work) <= _ROOT | {"reads", "taskLifecycle"}) _require(len(_json(work)) <= MAX_WORK_BYTES, "work_component_limit_exceeded") self._json_types(work) _require(work["schemaVersion"] == SCHEMA) _identifier(work["taskId"]) _sha(work["secret"]) _sha(work["mac"]) if "taskLifecycle" in work: self._validate_lifecycle(work["taskLifecycle"]) _fields(work["binding"], _BINDING) _require(work["binding"] == binding and work["binding"]["baseUrl"] == "https://concresca.com") for field in ("agentId", "workspaceId", "projectId"): _segment(work["binding"][field], field) _require(work["binding"]["workspaceId"] != work["binding"]["projectId"]) expected_mac = _mac(work, "state", {k: v for k, v in work.items() if k != "mac"}) _require(hmac.compare_digest(work["mac"], expected_mac)) ledger = _budget_call(budget.load_ledger, work["ledger"], scope_digest=_scope(work)) _require(work["halted"] in (None, "runtime_capacity_exceeded", "clock_invalid")) if work["continuityCredentialId"] is not None: _identifier(work["continuityCredentialId"]) intents = work["intents"] _require(type(intents) is list and len(intents) <= budget.MAX_OPERATIONS) ids, keys, attempts, unfinished = set(), set(), [], [] pool_counts = {"work": 0, "cleanup": 0} operations = {op["operationId"]: op for op in ledger["operations"]} for intent in intents: _require(type(intent) is dict and set(intent) in (_INTENT, _INTENT | {"resolutionBinding"})) if "resolutionBinding" in intent: _sha(intent["resolutionBinding"]) _identifier(intent["intentId"]) for field in ("requestDigest", "bindingDigest", "keyScopeDigest"): _sha(intent[field]) _require(intent["intentId"] not in ids and intent["keyScopeDigest"] not in keys) ids.add(intent["intentId"]) keys.add(intent["keyScopeDigest"]) _require(type(intent["pool"]) is str and intent["pool"] in pool_counts) pool_counts[intent["pool"]] += 1 _require(type(intent["attemptIds"]) is list and len(intent["attemptIds"]) <= budget.MAX_OPERATIONS) resolution_started = False for attempt in intent["attemptIds"]: _require(type(attempt) is str and attempt in operations) op = operations[attempt] resolution = self._resolution_attempt(work, intent, op) _require(not resolution_started or resolution, "work_state_invalid") resolution_started = resolution_started or resolution _require((resolution or op["bindingDigest"] == _mac(work, "attempt", [intent["bindingDigest"], attempt])) and op["pool"] == ("cleanup" if resolution else intent["pool"]) and op["units"] == self._units() and op["state"] in {"unknown", "reconciled"}) if op["state"] == "reconciled": _require(op["reconciliation"]["spent"] == self._units() and op["reconciliation"]["effectOutcome"] == "completed" and op["generation"] == 3) attempts.append(attempt) _require(intent["state"] in {"prepared", "unknown", "completed", "prevented"}) if intent["state"] in {"completed", "prevented"}: _sha(intent["resultDigest"]) _require(bool(intent["attemptIds"]) and operations[intent["attemptIds"][-1]]["state"] == "reconciled") _require(all(operations[a]["state"] == "unknown" for a in intent["attemptIds"][:-1])) reference = intent["resultReference"] _require(type(reference) is dict and bool(reference) and len(_json(reference)) <= MAX_REFERENCE_BYTES) last = operations[intent["attemptIds"][-1]] if self._resolution_attempt(work, intent, last): _require(reference.get("outcome") in {"completed", "prevented"} and intent["state"] == reference["outcome"], "work_state_invalid") else: _require(intent["state"] == "completed" and "resolutionBinding" not in intent) _require(intent["resultDigest"] == last["reconciliation"]["evidenceDigest"] == _mac(work, "response", [last["operationId"], last["bindingDigest"], reference])) else: _require(intent["resultDigest"] is None and intent["resultReference"] is None and bool(intent["attemptIds"]) == (intent["state"] == "unknown")) _require(all(operations[a]["state"] == "unknown" for a in intent["attemptIds"])) unfinished.append(intent) # Mutation order remains a subsequence of the one global attempt order. mutation_attempts = list(attempts) if "reads" in work: reads = work["reads"] _fields(reads, {"notBeforeMilliseconds", "attempts"}) _integer(reads["notBeforeMilliseconds"]) _require(type(reads["attempts"]) is list and len(reads["attempts"]) <= budget.MAX_OPERATIONS) read_attempts = [] for entry in reads["attempts"]: _fields(entry, {"operationId", "requestDigest", "pool", "checkpoint"}) _sha(entry["requestDigest"]) _require(type(entry["operationId"]) is str and entry["operationId"] in operations) op = operations[entry["operationId"]] _require(entry["pool"] in ("work", "cleanup") and op["pool"] == entry["pool"] and op["units"] == self._units() and op["state"] in {"unknown", "reconciled"} and op["bindingDigest"] == _mac(work, "read-attempt", [entry["operationId"], entry["requestDigest"], entry["pool"]])) if op["state"] == "unknown": _require(entry["checkpoint"] is None) else: checkpoint = self._read_checkpoint(entry["checkpoint"], binding) _require(op["generation"] == 3 and op["reconciliation"]["spent"] == self._units() and op["reconciliation"]["effectOutcome"] == "completed" and op["reconciliation"]["evidenceDigest"] == _mac(work, "read-response", [op["operationId"], op["bindingDigest"], checkpoint])) read_attempts.append(entry["operationId"]) _require(read_attempts == [a for a in operations if a in read_attempts]) attempts.extend(read_attempts) _require(len(attempts) == len(set(attempts)) and set(attempts) == set(operations)) _require(all(pool_counts[p] <= ledger["operationLimits"][p] for p in pool_counts)) _require(mutation_attempts == [a for a in operations if a in mutation_attempts] and list(operations) == ["attempt-%04d" % i for i in range(1, len(operations) + 1)]) pending = work["pending"] _require(len(unfinished) == (0 if pending is None else 1)) if pending is not None: _require(unfinished[0] == intents[-1]) _require(type(pending) is dict and set(pending) in (_PENDING, _PENDING | {"resolutionKey"})) _idempotency(pending["key"]) if _generation(work): _require(pending["key"] == _remote_key(work, "intent", [pending["intentId"], pending["method"], pending["path"]])) if "resolutionKey" in pending: _idempotency(pending["resolutionKey"]) _require(pending["resolutionKey"] != pending["key"]) self._resolution_descriptor(pending) _integer(pending["notBeforeMilliseconds"]) descriptor = self._descriptor(pending["method"], pending["path"], pending["body"]) intent = unfinished[0] _require(("resolutionKey" in pending) == ("resolutionBinding" in intent)) if "resolutionKey" in pending: _require(intent["resolutionBinding"] == _mac(work, "resolution", [intent["bindingDigest"], pending["resolutionKey"]])) if _generation(work): _require(pending["resolutionKey"] == _remote_key(work, "resolution", [intent["bindingDigest"]])) _require(pending["intentId"] == intent["intentId"] and intent["requestDigest"] == _mac(work, "request", [intent["intentId"], descriptor["method"], descriptor["path"], descriptor["body"]]) and pending["bindingDigest"] == intent["bindingDigest"] == _intent_binding(work, intent["intentId"], descriptor, pending["key"]) and intent["pool"] == _pool(descriptor) and intent["keyScopeDigest"] == _mac(work, "remote-key", [descriptor["method"], descriptor["path"], pending["key"]])) return copy.deepcopy(work) @staticmethod def _units(): return {"requests": 1, "runtimeMilliseconds": ATTEMPT_RUNTIME_MS} def _active(self, state, work, now): _require(state["phase"] == "active", "work_credential_not_active") _require(state.get("pending") is None, "work_continuity_pending") principal = self._agent._binding(state["principal"], state["credential"], state) expiry = datetime.fromisoformat(principal["credentialExpiresAt"].replace("Z", "+00:00")) delta = expiry - datetime(1970, 1, 1, tzinfo=timezone.utc) expiry_ms = (delta.days * 86400 + delta.seconds) * 1000 + delta.microseconds // 1000 _require(now < expiry_ms, "work_continuity_required") _require(not work or work["continuityCredentialId"] != principal["credentialId"], "work_continuity_required") return principal def _preflight(self, session, work, *, dispatch_headroom=False, revision_offset=1, read_headroom=False): """Validate/size a detached next state; never change or save session state.""" work["mac"] = _mac(work, "state", {k: v for k, v in work.items() if k != "mac"}) candidate = copy.deepcopy(session.state) candidate[STATE_KEY] = copy.deepcopy(work) self._validate(candidate) # Account for envelope overhead before changing caller/session state. envelope = copy.deepcopy(session.envelope) envelope.update(state=candidate, revision=envelope["revision"] + revision_offset, previous="0" * 64) completion = MAX_READ_CHECKPOINT_BYTES + 2048 if read_headroom else (COMPLETION_RESERVE_BYTES if dispatch_headroom else 0) _require(len(_json(work)) + completion <= MAX_WORK_BYTES, "work_component_limit_exceeded") _require(len(_json(envelope)) + completion + LIFECYCLE_RESERVE_BYTES <= _MAX_BYTES, "work_lifecycle_headroom_required") return candidate def _save(self, session, work, *, dispatch_headroom=False, read_headroom=False): session.state = self._preflight(session, work, dispatch_headroom=dispatch_headroom, read_headroom=read_headroom) session.save() # Returns only after both protected copies read back. def _summary(self, work, state): # Current local authority only; no /me, rotation or repair side effects. condition = None try: self._active(state, work, _integer(self._clock())) except AgentError as exc: condition = exc.code return {"schemaVersion": SCHEMA, "configured": True, "taskGeneration": _generation(work), "expectedGenerationRequired": _generation(work) > 0, "retiredAccounting": copy.deepcopy(work.get("taskLifecycle", {}).get("retired", _retired_zero())), "pending": work["pending"] is not None, "completedIntents": sum(i["state"] == "completed" for i in work["intents"]), "preventedIntents": sum(i["state"] == "prevented" for i in work["intents"]), "resolutionPending": work["pending"] is not None and "resolutionKey" in work["pending"], "retainedIntents": len(work["intents"]), "readAttempts": len(work.get("reads", {}).get("attempts", [])), "completedReads": sum(r["checkpoint"] is not None for r in work.get("reads", {}).get("attempts", [])), "unknownReads": sum(r["checkpoint"] is None for r in work.get("reads", {}).get("attempts", [])), "readNotBeforeMilliseconds": work.get("reads", {}).get("notBeforeMilliseconds", 0), "totals": _budget_call(budget.totals, work["ledger"], scope_digest=_scope(work)), "notBeforeMilliseconds": (work["pending"] or {}).get("notBeforeMilliseconds", 0), "continuityRequired": condition is not None, "continuityCondition": condition, "prior401EvidenceRetained": work["continuityCredentialId"] is not None, "workComponentLimitBytes": MAX_WORK_BYTES, "lifecycleExpansionReserveBytes": LIFECYCLE_RESERVE_BYTES, "arbitraryRecoveryChainHeadroomGuaranteed": False, "halted": work["halted"], "attemptRuntimeReservationMilliseconds": ATTEMPT_RUNTIME_MS, "runtimeAccounting": "full_reserved_attempt_capacity_not_measured_usage", "budgetScope": "selected_task", "additionalTaskConfigurationSupported": True, "runtimeWallClockEnforced": False, "remoteResourceEnforced": False, "standardWorkTransportConfigured": self._agent._factory is AutonomousCommonsClient, "standardWorkAbortAfterMilliseconds": ATTEMPT_RUNTIME_MS - _WORK_TEARDOWN_MS, "transportAcceptanceVerifiedByStatus": False, "valuesRedacted": True, "rawCredentialExposed": False, "rawPayloadExposed": False, "judgmentState": "NONE"} @_redacted def configure_budget(self, *, task_id, limits, cleanup_reserve, deadline_ms, operation_limit=budget.MAX_OPERATIONS, cleanup_operation_slots=8): """Select once. An exact repeated configuration is read-only; no reset.""" _identifier(task_id) with self._agent._custody.transaction() as session: self._agent._validate_state(session.state) existing = STATE_KEY in session.state work = self._validate(session.state) if existing else { "schemaVersion": SCHEMA, "taskId": task_id, "binding": copy.deepcopy(session.state["binding"]), "secret": secrets.token_hex(32), "ledger": None, "intents": [], "pending": None, "continuityCredentialId": None, "halted": None, "mac": "0" * 64} selected = _budget_call(budget.create_ledger, scope_digest=_scope(work), limits=limits, cleanup_reserve=cleanup_reserve, deadline_ms=deadline_ms, operation_limit=operation_limit, cleanup_operation_slots=cleanup_operation_slots) if existing: _require(task_id == work["taskId"] and all(work["ledger"][k] == selected[k] for k in ("limits", "cleanupReserve", "deadlineMilliseconds", "operationLimits")), "work_budget_already_configured") else: self._active(session.state, None, _integer(self._clock())) work["ledger"] = selected self._save(session, work) return self._summary(work, session.state) @_redacted def advance_task(self, expected_generation, next_task_id, *, limits, cleanup_reserve, deadline_ms, operation_limit=budget.MAX_OPERATIONS, cleanup_operation_slots=8): """Atomically compact a terminal task and select a generation-bound next one. Retired totals are permanent historical exposure, not a bank, refund, identity-lifetime allowance or independently reconstructible history. No remote agreement is completed, cancelled or withdrawn by this call. """ _integer(expected_generation) _identifier(next_task_id) with self._agent._custody.transaction() as session: work = self._validate(session.state) # Pure validation before any counter changes; exact repeated # configurations have one canonical transition identity. configured = _budget_call(budget.create_ledger, scope_digest=_scope(work), limits=limits, cleanup_reserve=cleanup_reserve, deadline_ms=deadline_ms, operation_limit=operation_limit, cleanup_operation_slots=cleanup_operation_slots) configuration = {k: configured[k] for k in ("limits", "cleanupReserve", "deadlineMilliseconds", "operationLimits")} advance_digest = _mac(work, "lifecycle-advance", [work["binding"], expected_generation, next_task_id, configuration]) generation = _generation(work) if generation == expected_generation + 1: _require(work["taskLifecycle"]["lastAdvanceDigest"] == advance_digest, "work_task_advance_conflict") # A duplicate is only inspection, even after successor progress, # expiry or credential changes. Never restore its initial ledger. return self._summary(work, session.state) _check_generation(work, expected_generation) _require(work["halted"] is None, "work_runtime_halted") self._active(session.state, work, _integer(self._clock())) _require(work["pending"] is None and all(i["state"] in {"completed", "prevented"} for i in work["intents"]), "work_task_not_terminal") successor_generation = _add_counter(generation, 1) retired = copy.deepcopy(work.get("taskLifecycle", {}).get("retired", _retired_zero())) retired["tasks"] = _add_counter(retired["tasks"], 1) for field, amount in ( ("completedIntents", sum(i["state"] == "completed" for i in work["intents"])), ("preventedIntents", sum(i["state"] == "prevented" for i in work["intents"])), ("completedReads", sum(r["checkpoint"] is not None for r in work.get("reads", {}).get("attempts", []))), ("unknownReads", sum(r["checkpoint"] is None for r in work.get("reads", {}).get("attempts", [])))): retired[field] = _add_counter(retired[field], amount) for operation in work["ledger"]["operations"]: category = "spent" if operation["state"] == "reconciled" else "unknown" accumulator = retired["pools"][operation["pool"]][category] accumulator["attempts"] = _add_counter(accumulator["attempts"], 1) for unit in budget.UNITS: accumulator[unit] = _add_counter(accumulator[unit], operation["units"][unit]) commitment = _mac(work, "lifecycle-retirement", {k: v for k, v in work.items() if k not in {"secret", "mac"}}) previous = work.get("taskLifecycle", {}).get("chainHead") or _mac(work, "lifecycle-genesis", work["binding"]) chain = _mac(work, "lifecycle-chain", [previous, commitment, advance_digest, retired]) work["taskLifecycle"] = {"generation": successor_generation, "retired": retired, "chainHead": chain, "lastAdvanceDigest": advance_digest} work.update(taskId=next_task_id, intents=[], pending=None) if "reads" in work: # Retry barriers survive; observation bodies/references/cursors do # not become a historical body journal or new cleanup authority. work["reads"] = {"notBeforeMilliseconds": work["reads"]["notBeforeMilliseconds"], "attempts": []} work["ledger"] = _budget_call(budget.create_ledger, scope_digest=_scope(work), limits=limits, cleanup_reserve=cleanup_reserve, deadline_ms=deadline_ms, operation_limit=operation_limit, cleanup_operation_slots=cleanup_operation_slots) self._save(session, work) return self._summary(work, session.state) @_redacted def prepare(self, *, intent_id, method, path, body, idempotency_key=None, expected_generation=None): """Seal exact POST intent without sending; repeated IDs never redispatch.""" _identifier(intent_id) if idempotency_key is not None: _idempotency(idempotency_key) descriptor = self._descriptor(method, path, body) with self._agent._custody.transaction() as session: work = self._validate(session.state) _check_generation(work, expected_generation) _require(not _generation(work) or idempotency_key is None, "work_generation_key_override_forbidden") request_digest = _mac(work, "request", [intent_id, method, path, descriptor["body"]]) for existing in work["intents"]: if existing["intentId"] == intent_id: _require(existing["requestDigest"] == request_digest, "work_intent_conflict") if idempotency_key is not None: _require(existing["bindingDigest"] == _intent_binding(work, intent_id, descriptor, idempotency_key), "work_intent_conflict") return self._summary(work, session.state) if idempotency_key is None: idempotency_key = _remote_key(work, "intent", [intent_id, method, path]) digest = _intent_binding(work, intent_id, descriptor, idempotency_key) _require(work["halted"] is None, "work_runtime_halted") self._active(session.state, work, _integer(self._clock())) _require(work["pending"] is None, "work_intent_pending") pool = _pool(descriptor) _require(sum(i["pool"] == pool for i in work["intents"]) < work["ledger"]["operationLimits"][pool], "work_intent_limit_exceeded") key_digest = _mac(work, "remote-key", [method, path, idempotency_key]) _require(all(i["keyScopeDigest"] != key_digest for i in work["intents"]), "work_key_reused") # Size the exact first physical attempt, but do not persist its # reservation. Later resumes still recheck capacity and authority. operation_id = "attempt-%04d" % (len(work["ledger"]["operations"]) + 1) attempt_digest = _mac(work, "attempt", [digest, operation_id]) admission_ledger = _budget_call(budget.reserve, work["ledger"], scope_digest=_scope(work), operation_id=operation_id, binding_digest=attempt_digest, units=self._units(), now_ms=_integer(self._clock()), pool=pool) work["intents"].append({"intentId": intent_id, "requestDigest": request_digest, "bindingDigest": digest, "keyScopeDigest": key_digest, "pool": pool, "attemptIds": [], "state": "prepared", "resultDigest": None, "resultReference": None}) work["pending"] = {"intentId": intent_id, "method": method, "path": path, "body": copy.deepcopy(descriptor["body"]), "key": idempotency_key, "bindingDigest": digest, "notBeforeMilliseconds": 0} # Every admitted request must have an exact bounded resolution wire # representation. This check sends nothing and reserves no resources. self._resolution_descriptor(work["pending"]) projected = copy.deepcopy(work) projected["ledger"] = _budget_call(budget.mark_unknown, admission_ledger, scope_digest=_scope(work), operation_id=operation_id, binding_digest=attempt_digest) projected["intents"][-1].update(state="unknown", attemptIds=[operation_id]) # Prepare + first dispatch require two envelope revisions. This # allowance is not a promise that future retries/lifecycle growth fit. self._preflight(session, projected, dispatch_headroom=True, revision_offset=2) self._save(session, work) return self._summary(work, session.state) @_redacted def status(self): """No network/lifecycle calls. Existing custody load may heal its peer.""" with self._agent._custody.transaction() as session: self._agent._validate_state(session.state) if STATE_KEY not in session.state: return {"schemaVersion": SCHEMA, "configured": False, "valuesRedacted": True} return self._summary(self._validate(session.state), session.state) @_redacted def read(self, path, *, cleanup_for=None, expected_generation=None): """One metered current GET; never resume or replace a pending mutation. Only the compact observation checkpoint survives in sealed custody. Saving it proves observation, NOT delivery/consumption by this caller. Every explicit call is a fresh charged attempt, including a repeated URL. """ descriptor = self._read_descriptor(path) with self._agent._custody.transaction() as session: work = self._validate(session.state) _check_generation(work, expected_generation) _require(work["halted"] is None, "work_runtime_halted") now = _integer(self._clock()) principal = self._active(session.state, work, now) reads = work.setdefault("reads", {"notBeforeMilliseconds": 0, "attempts": []}) _require(now >= reads["notBeforeMilliseconds"], "work_retry_deferred") pool = self._read_pool(work, path, cleanup_for) operation_id = "attempt-%04d" % (len(work["ledger"]["operations"]) + 1) request_digest = _mac(work, "read-request", [work["taskId"], work["binding"], "GET", path, pool, cleanup_for]) binding_digest = _mac(work, "read-attempt", [operation_id, request_digest, pool]) scope = _scope(work) client = self._agent._client(session.state["credential"]) client._bearer() # No anonymous fallback, refresh or separate /me request. work["ledger"] = _budget_call(budget.reserve, work["ledger"], scope_digest=scope, operation_id=operation_id, binding_digest=binding_digest, units=self._units(), now_ms=now, pool=pool) work["ledger"] = _budget_call(budget.mark_unknown, work["ledger"], scope_digest=scope, operation_id=operation_id, binding_digest=binding_digest) entry = {"operationId": operation_id, "requestDigest": request_digest, "pool": pool, "checkpoint": None} reads["attempts"].append(entry) self._save(session, work, read_headroom=True) dispatch_now = _integer(self._clock()) self._active(session.state, work, dispatch_now) _require(dispatch_now >= reads["notBeforeMilliseconds"], "work_retry_deferred") _require(pool == "cleanup" or dispatch_now < work["ledger"]["deadlineMilliseconds"], "work_deadline_reached") started = _integer(self._monotonic(), maximum=(1 << 63) - 1) failure, result = None, None try: with self._agent._work_transport(session, client, started, ATTEMPT_RUNTIME_MS) as transport: result = self._adapter().read_once(transport, copy.deepcopy(descriptor), copy.deepcopy(work["binding"])) _fields(result, {"observation", "checkpoint"}) checkpoint = self._read_checkpoint(result["checkpoint"], work["binding"]) if _READ_OBJECT.fullmatch(path): _require(checkpoint.get("resourceId") == path.rsplit("/", 1)[1], "work_read_checkpoint_invalid") else: bare = path.split("?", 1)[0] proposal_page = bare.endswith("/proposals") _require("itemIds" in checkpoint and checkpoint["resourceKind"] == ("work_proposal" if proposal_page else "work_opportunity") and checkpoint["opportunityId"] == (bare.split("/")[-2] if proposal_page else None), "work_read_checkpoint_invalid") observation = result["observation"] _require(type(observation) is dict, "work_read_observation_invalid") _require(observation.get("schemaVersion") == "concresca.work_read_observation.v1", "work_read_observation_invalid") # Transient content has its own wire-sized bound, never custody's # 128 KiB serializer or the body-free checkpoint allocation. self._json_types(observation, depth=1) _require(len(json.dumps(observation, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False).encode("utf-8")) <= MAX_READ_OBSERVATION_BYTES, "work_read_observation_invalid") except Exception as exc: failure = exc ended = self._monotonic() if type(ended) is not int or ended < started or ended > (1 << 63) - 1: work["halted"] = "clock_invalid" elif ended - started > ATTEMPT_RUNTIME_MS * 1_000_000: work["halted"] = "runtime_capacity_exceeded" if work["halted"]: self._save(session, work) raise AgentError("work_runtime_halted") if failure is not None: if isinstance(failure, CommonsClientError): retry = failure.retry_after_seconds if type(retry) is int and 1 <= retry <= 86400: reads["notBeforeMilliseconds"] = max(reads["notBeforeMilliseconds"], _integer(self._clock()) + retry * 1000) if type(failure.http_status) is int and failure.http_status == 401: work["continuityCredentialId"] = principal["credentialId"] self._save(session, work) raise AgentError("work_continuity_required" if work["continuityCredentialId"] == principal["credentialId"] else "work_read_attempt_unknown") from None op = work["ledger"]["operations"][-1] evidence = _mac(work, "read-response", [operation_id, binding_digest, checkpoint]) verification = {"schemaVersion": budget.VERIFICATION_SCHEMA, "verified": True, "scopeDigest": scope, "operationId": operation_id, "bindingDigest": binding_digest, "pool": pool, "reservationDigest": op["reservationDigest"], "effectOutcome": "completed", "spent": self._units(), "evidenceDigest": evidence} work["ledger"] = _budget_call(budget.reconcile_verified, work["ledger"], scope_digest=scope, operation_id=operation_id, binding_digest=binding_digest, verification=verification) entry["checkpoint"] = checkpoint self._save(session, work) return {**self._summary(work, session.state), "observation": copy.deepcopy(observation), "checkpoint": copy.deepcopy(checkpoint), "resultSource": "fresh_validated_get_observation", "consumptionAcknowledged": False, "containsSelectedPublicPayload": True, "rawPayloadExposed": True, "valuesRedacted": False} @_redacted def resume(self, *, expected_generation=None): """At most one physical attempt; exceptions never release its exposure.""" with self._agent._custody.transaction() as session: work = self._validate(session.state) _check_generation(work, expected_generation) return self._resume_locked(session, work) @_redacted def resolve(self, *, expected_generation=None): """Reconcile-or-prevent the pending mutation; one cleanup attempt. Resolution is sticky across restarts. A local cancellation alone never proves that a remote effect was absent or prevents a delayed execution. """ with self._agent._custody.transaction() as session: work = self._validate(session.state) _check_generation(work, expected_generation) pending = work["pending"] if pending is not None and "resolutionKey" not in pending: _require(work["halted"] is None, "work_runtime_halted") now = _integer(self._clock()) self._active(session.state, work, now) self._resolution_descriptor(pending) pending["resolutionKey"] = _remote_key(work, "resolution", [work["intents"][-1]["bindingDigest"]]) _require(pending["resolutionKey"] != pending["key"], "work_key_reused") work["intents"][-1]["resolutionBinding"] = _mac(work, "resolution", [work["intents"][-1]["bindingDigest"], pending["resolutionKey"]]) projected = copy.deepcopy(work) intent = projected["intents"][-1] operation_id = "attempt-%04d" % (len(projected["ledger"]["operations"]) + 1) binding_digest = _mac(projected, "resolution-attempt", [intent["resolutionBinding"], operation_id]) projected["ledger"] = _budget_call(budget.reserve, projected["ledger"], scope_digest=_scope(projected), operation_id=operation_id, binding_digest=binding_digest, units=self._units(), now_ms=now, pool="cleanup") projected["ledger"] = _budget_call(budget.mark_unknown, projected["ledger"], scope_digest=_scope(projected), operation_id=operation_id, binding_digest=binding_digest) intent["attemptIds"].append(operation_id) intent["state"] = "unknown" # Fail admission without changing pending mode if the first # resolution attempt and its completion could not fit. self._preflight(session, projected, dispatch_headroom=True, revision_offset=2) # Sticky mode requires a durable copy. If this save fails before # either copy is written, explicitly resolve again; blind resume # still means the original mutation. A partial save heals sticky. self._save(session, work) return self._resume_locked(session, work) def _resume_locked(self, session, work): pending = work["pending"] if pending is None: reference = work["intents"][-1]["resultReference"] if work["intents"] else None return {**self._summary(work, session.state), "result": copy.deepcopy(reference), "resultSource": "sealed_last_validated_response_not_current_readback"} _require(work["halted"] is None, "work_runtime_halted") now = _integer(self._clock()) principal = self._active(session.state, work, now) _require(now >= pending["notBeforeMilliseconds"], "work_retry_deferred") descriptor = self._descriptor(pending["method"], pending["path"], pending["body"]) resolving = "resolutionKey" in pending if resolving: self._resolution_descriptor(pending) intent = next(i for i in work["intents"] if i["intentId"] == pending["intentId"]) operation_id = "attempt-%04d" % (len(work["ledger"]["operations"]) + 1) binding_digest = _mac(work, "resolution-attempt" if resolving else "attempt", [intent["resolutionBinding"] if resolving else intent["bindingDigest"], operation_id]) pool = "cleanup" if resolving else intent["pool"] scope = _scope(work) client = self._agent._client(session.state["credential"]) work["ledger"] = _budget_call(budget.reserve, work["ledger"], scope_digest=scope, operation_id=operation_id, binding_digest=binding_digest, units=self._units(), now_ms=now, pool=pool) work["ledger"] = _budget_call(budget.mark_unknown, work["ledger"], scope_digest=scope, operation_id=operation_id, binding_digest=binding_digest) intent["attemptIds"].append(operation_id) intent["state"] = "unknown" self._save(session, work, dispatch_headroom=True) # Recheck retry timing too: the wall clock can move back during a save. # Retain the already saved unknown charge even when dispatch is deferred. dispatch_now = _integer(self._clock()) self._active(session.state, work, dispatch_now) _require(dispatch_now >= pending["notBeforeMilliseconds"], "work_retry_deferred") _require(pool == "cleanup" or dispatch_now < work["ledger"]["deadlineMilliseconds"], "work_deadline_reached") started = _integer(self._monotonic(), maximum=(1 << 63) - 1) failure, reference = None, None try: with self._agent._work_transport(session, client, started, ATTEMPT_RUNTIME_MS) as transport: if resolving: reference = self._adapter().send_resolution_once(transport, copy.deepcopy(descriptor), copy.deepcopy(work["binding"]), pending["key"], pending["resolutionKey"]) _require(type(reference) is dict and reference.get("outcome") in {"completed", "prevented"}, "work_response_reference_invalid") else: reference = self._adapter().send_once(transport, copy.deepcopy(descriptor), copy.deepcopy(work["binding"]), pending["key"]) self._json_types(reference) _require(type(reference) is dict and bool(reference) and len(_json(reference)) <= MAX_REFERENCE_BYTES, "work_response_reference_invalid") except Exception as exc: failure = exc ended = self._monotonic() if type(ended) is not int or ended < started or ended > (1 << 63) - 1: work["halted"] = "clock_invalid" elif ended - started > ATTEMPT_RUNTIME_MS * 1_000_000: work["halted"] = "runtime_capacity_exceeded" if work["halted"]: self._save(session, work) raise AgentError("work_runtime_halted") if failure is not None: if isinstance(failure, CommonsClientError): retry = failure.retry_after_seconds if type(retry) is int and 1 <= retry <= 86400: pending["notBeforeMilliseconds"] = _integer(self._clock()) + retry * 1000 if type(failure.http_status) is int and failure.http_status == 401: work["continuityCredentialId"] = principal["credentialId"] self._save(session, work) raise AgentError("work_continuity_required" if work["continuityCredentialId"] == principal["credentialId"] else "work_attempt_unknown") from None op = work["ledger"]["operations"][-1] evidence = _mac(work, "response", [operation_id, binding_digest, reference]) verification = {"schemaVersion": budget.VERIFICATION_SCHEMA, "verified": True, "scopeDigest": scope, "operationId": operation_id, "bindingDigest": binding_digest, "pool": pool, "reservationDigest": op["reservationDigest"], "effectOutcome": "completed", "spent": self._units(), "evidenceDigest": evidence} work["ledger"] = _budget_call(budget.reconcile_verified, work["ledger"], scope_digest=scope, operation_id=operation_id, binding_digest=binding_digest, verification=verification) intent.update(state="prevented" if resolving and reference["outcome"] == "prevented" else "completed", resultDigest=evidence, resultReference=copy.deepcopy(reference)) work["pending"] = None self._save(session, work) # Reference is explicitly adapter-validated, not participant content. return {**self._summary(work, session.state), "result": copy.deepcopy(reference), "resultSource": "sealed_last_validated_response_not_current_readback"} @_redacted def validate_work_state(work_state, binding): """Pure optional-field validator for AutonomousAgent._validate_state. Caller invokes only when STATE_KEY exists. Does not call back into the agent, open custody, inspect credentials, or send. Returns detached validated state. """ return WorkController(None)._validate_work(work_state, binding) def configure(agent, task_id, *, limits, cleanup_reserve, deadline_ms, operation_limit=budget.MAX_OPERATIONS, cleanup_operation_slots=8): return WorkController(agent).configure_budget(task_id=task_id, limits=limits, cleanup_reserve=cleanup_reserve, deadline_ms=deadline_ms, operation_limit=operation_limit, cleanup_operation_slots=cleanup_operation_slots) def prepare(agent, intent_id, method, path, body, *, idempotency_key=None, expected_generation=None): return WorkController(agent).prepare(intent_id=intent_id, method=method, path=path, body=body, idempotency_key=idempotency_key, expected_generation=expected_generation) def resume(agent, *, expected_generation=None): return WorkController(agent).resume(expected_generation=expected_generation) def resolve(agent, *, expected_generation=None): return WorkController(agent).resolve(expected_generation=expected_generation) def status(agent): return WorkController(agent).status() def read_work(agent, path, cleanup_for=None, *, expected_generation=None): return WorkController(agent).read(path, cleanup_for=cleanup_for, expected_generation=expected_generation) def advance_task(agent, expected_generation, next_task_id, *, limits, cleanup_reserve, deadline_ms, operation_limit=budget.MAX_OPERATIONS, cleanup_operation_slots=8): return WorkController(agent).advance_task(expected_generation, next_task_id, limits=limits, cleanup_reserve=cleanup_reserve, deadline_ms=deadline_ms, operation_limit=operation_limit, cleanup_operation_slots=cleanup_operation_slots)