"""Candidate command service with one root-level consistency boundary.""" from __future__ import annotations import asyncio from hashlib import sha256 from typing import Any from weakref import WeakValueDictionary from cyber_agent.application.candidate import ( CandidateAdoptionReceipt, CandidateAdoptionRequest, CandidateLedger, CandidateLifecycleRecord, CandidateOperationRecord, CandidatePointer, CandidateRef, CandidateRepository, CandidateRepositoryRejected, CandidateReviewAction, CandidateVersionRequest, MAX_CANDIDATES_PER_ROOT, MAX_CANDIDATE_OPERATIONS_PER_ROOT, ) from cyber_agent.application.models import canonical_json from cyber_agent.application.quality import ( CandidateValidationCheckpoint, CandidateValidationRecord, ) from cyber_agent.core.agent_mode import require_mutable_trace_policy from cyber_agent.core.artifacts import ArtifactRef, ArtifactResolver, resolve_artifact_refs from cyber_agent.core.task_protocol import ensure_task_protocol, task_contract_ref from cyber_agent.tools.errors import RecoverableToolExecutionError class CandidateStateError(ValueError): pass class CandidateService: """Validate, persist, and recover candidate version operations.""" _locks: "WeakValueDictionary[tuple[asyncio.AbstractEventLoop, int, str], asyncio.Lock]" = WeakValueDictionary() def __init__( self, *, store: Any, application_binding: Any, repository: CandidateRepository, artifact_resolver: ArtifactResolver, event_service: Any = None, ) -> None: self.store = store self.binding = application_binding self.repository = repository self.artifact_resolver = artifact_resolver self.event_service = event_service def _lock_for(self, root_trace_id: str) -> asyncio.Lock: key = (asyncio.get_running_loop(), id(self.store), root_trace_id) lock = self._locks.get(key) if lock is None: lock = asyncio.Lock() self._locks[key] = lock return lock async def has_replayable_version_command( self, trace_id: str, *, command_id: str, arguments: dict[str, Any], ) -> bool: """Verify the exact durable candidate command behind an orphaned call.""" try: request = arguments.get("request") if not isinstance(request, dict): return False operation = request.get("operation") if operation not in {"create", "fork", "merge"}: return False parent_refs = [ CandidatePointer.model_validate(item) for item in request.get("parent_refs", []) ] input_hash = sha256(canonical_json({ "operation": operation, "content": request.get("content"), "parent_refs": [ item.model_dump(mode="json") for item in parent_refs ], }).encode("utf-8")).hexdigest() command_hash = sha256(canonical_json({ "trace_id": trace_id, "command_id": command_id, }).encode("utf-8")).hexdigest() operation_id = f"candidate:{command_hash}" _trace, root = await self._require_owner_trace(trace_id) ledger = await self._load_ledger(root.trace_id) except Exception: return False return any( item.operation_id == operation_id and item.command_id == command_id and item.operation == operation and item.input_hash == input_hash and item.status in {"pending", "completed", "failed"} for item in ledger.operations ) async def manage( self, trace_id: str, *, operation: str, content: Any, parent_refs: list[CandidatePointer], effective_at_sequence: int, command_id: str | None = None, ) -> CandidateRef: trace, root = await self._require_owner_trace(trace_id) root_trace_id = root.trace_id input_payload = { "operation": operation, "content": content, "parent_refs": [item.model_dump(mode="json") for item in parent_refs], } input_hash = sha256( canonical_json(input_payload).encode("utf-8") ).hexdigest() if command_id is not None and (not command_id or len(command_id) > 300): raise CandidateStateError( "Candidate command_id must be a non-empty string up to 300 chars" ) # A persisted tool_call_id identifies one durable command. Direct # internal callers that do not have an assistant call identity retain # a sequence-scoped fallback, but the public tool always supplies it. command_identity = command_id or f"sequence:{effective_at_sequence}" command_hash = sha256( canonical_json({ "trace_id": trace_id, "command_id": command_identity, }).encode("utf-8") ).hexdigest() operation_id = f"candidate:{command_hash}" async with self._lock_for(root_trace_id): ledger = await self._load_ledger(root_trace_id) existing_operation = next( ( item for item in ledger.operations if item.operation_id == operation_id ), None, ) recovering_pending_operation = existing_operation is not None if existing_operation is not None: if existing_operation.input_hash != input_hash: raise CandidateStateError( "Candidate operation payload changed during recovery" ) if existing_operation.status == "completed": if existing_operation.candidate is None: raise CandidateStateError( "Completed candidate operation has no result" ) return ledger.candidate(existing_operation.candidate) if existing_operation.status == "failed": raise CandidateStateError( existing_operation.error or "Candidate operation failed" ) if self._reserved_candidate_slots(ledger) > MAX_CANDIDATES_PER_ROOT: raise CandidateStateError( "Candidate ledger candidate reservations are corrupt" ) else: # Every supported version command publishes exactly one new # immutable CandidateRef. Reject a full aggregate before we # persist an intent or call the application Repository; doing # this only in CandidateLedger validation would leave an # external orphan after the Repository has committed. if self._reserved_candidate_slots(ledger) >= MAX_CANDIDATES_PER_ROOT: raise CandidateStateError( "Candidate ledger candidate limit exceeded before execution" ) if len(ledger.operations) >= MAX_CANDIDATE_OPERATIONS_PER_ROOT: raise CandidateStateError( "Candidate operation limit exceeded before execution" ) try: pointer = self._allocate_pointer( operation, trace_id, effective_at_sequence, parent_refs, ledger, ) except ValueError as exc: raise CandidateStateError(str(exc)) from exc operation_record = CandidateOperationRecord( operation_id=operation_id, command_id=command_id, operation=operation, input_hash=input_hash, candidate=pointer, ) ledger = ledger.model_copy(update={ "operations": (*ledger.operations, operation_record), }) await self._persist_ledger(root_trace_id, ledger) existing_operation = operation_record assert existing_operation.candidate is not None pointer = existing_operation.candidate try: parents = [ledger.candidate(item) for item in parent_refs] self._validate_parents(trace, root, parents) for parent in parents: await self._assert_candidate_active(ledger, parent) state = ensure_task_protocol(trace.context) request = CandidateVersionRequest( operation_id=operation_id, application_ref=self.binding.application_ref, root_trace_id=root_trace_id, owner_trace_id=trace_id, uid=trace.uid, candidate_id=pointer.candidate_id, revision=pointer.revision, task_contract_ref=task_contract_ref( state, root_task_anchor_hash=trace.context.get( "root_task_anchor_hash" ), ), parent_refs=tuple(parent_refs), content=content, ) try: raw_artifact_ref = ( await self.repository.merge(request) if operation == "merge" else await self.repository.put_version(request) ) except CandidateRepositoryRejected as exc: raise CandidateStateError(str(exc)) from exc except Exception as exc: raise RecoverableToolExecutionError( "Candidate Repository execution outcome is unknown; " "the original idempotent command must be replayed" ) from exc try: artifact_ref = ArtifactRef.model_validate(raw_artifact_ref) except ValueError as exc: raise CandidateStateError( f"CandidateRepository returned an invalid ArtifactRef: {exc}" ) from exc materials, issues = await resolve_artifact_refs( [artifact_ref], resolver=self.artifact_resolver, root_trace_id=root_trace_id, uid=trace.uid, ) if issues or len(materials) != 1: reason = issues[0].reason if issues else "Artifact was not resolved" if any(issue.outcome == "unknown" for issue in issues): raise RecoverableToolExecutionError( "Candidate ArtifactResolver is temporarily unavailable; " "the original command must remain replayable" ) raise CandidateStateError( f"CandidateRepository returned an invalid ArtifactRef: {reason}" ) candidate_ref = CandidateRef( **pointer.model_dump(mode="json"), application_ref=self.binding.application_ref, root_trace_id=root_trace_id, owner_trace_id=trace_id, task_contract_ref=request.task_contract_ref, artifact_ref=artifact_ref, parent_refs=tuple(parent_refs), created_at_sequence=effective_at_sequence, ) except Exception as exc: if isinstance(exc, RecoverableToolExecutionError): raise # A recovered pending intent may already have committed in the # application Repository. A transient Repository/Resolver error # cannot prove failure, so keep the original command replayable. if ( recovering_pending_operation and not isinstance(exc, CandidateStateError) ): raise RecoverableToolExecutionError( "Candidate recovery was interrupted before its durable " "ledger outcome could be established" ) from exc failed = self._fail_operation(ledger, operation_id, str(exc)) try: await self._persist_ledger(root_trace_id, failed) except Exception as publish_exc: raise RecoverableToolExecutionError( "Candidate Repository result is durable but its failed " "ledger outcome must be replayed with the original command ID" ) from publish_exc raise completed = self._complete_operation( ledger, operation_id, candidate_ref, ) # If this atomic publish fails after the Repository committed, the # pending operation remains recoverable. Re-execution uses the same # operation_id and the Repository's idempotency contract. try: await self._persist_ledger(root_trace_id, completed) except Exception as exc: raise RecoverableToolExecutionError( "Candidate Repository result is durable but its ledger publish " "must be replayed with the original command ID" ) from exc await self._emit_candidate_registered(candidate_ref, operation_id) return candidate_ref async def validate_report_refs( self, trace_id: str, refs: list[CandidateRef], *, active_head_sequence: int | None = None, ) -> None: trace, root = await self._require_owner_trace(trace_id) async with self._lock_for(root.trace_id): ledger = await self._load_ledger(root.trace_id) for ref in refs: persisted = ledger.candidate(ref) if persisted != ref: raise CandidateStateError( "TaskReport CandidateRef does not match the root ledger" ) self._validate_candidate_owner(persisted, trace, root) await self._assert_candidate_active( ledger, persisted, active_head_sequence=active_head_sequence, ) if await self._current_state( ledger, ref, active_trace_id=trace_id, active_head_sequence=active_head_sequence, ) != "proposed": raise CandidateStateError( "TaskReport may only reference proposed candidates" ) async def resolve_for_validation( self, trace_id: str, candidate_ref: CandidateRef, ): """Resolve one exact report-owned candidate through the trusted resolver.""" trace, root = await self._require_bound_trace(trace_id) async with self._lock_for(root.trace_id): ledger = await self._load_ledger(root.trace_id) persisted = ledger.candidate(candidate_ref) if persisted != candidate_ref: raise CandidateStateError( "CandidateRef does not match the root ledger" ) self._validate_candidate_owner(persisted, trace, root) await self._assert_candidate_active(ledger, persisted) materials, issues = await resolve_artifact_refs( [candidate_ref.artifact_ref], resolver=self.artifact_resolver, root_trace_id=root.trace_id, uid=trace.uid, ) if issues or len(materials) != 1: reason = issues[0].reason if issues else "Artifact was not resolved" raise CandidateStateError( f"Candidate ArtifactRef cannot be validated: {reason}" ) return materials[0] async def cached_validation( self, trace_id: str, candidate_ref: CandidateRef, plan_hash: str, ) -> CandidateValidationRecord | None: trace, root = await self._require_bound_trace(trace_id) async with self._lock_for(root.trace_id): ledger = await self._load_ledger(root.trace_id) persisted = ledger.candidate(candidate_ref) self._validate_candidate_owner(persisted, trace, root) await self._assert_candidate_active(ledger, persisted) for raw in ledger.validations: record = CandidateValidationRecord.model_validate(raw) if ( record.candidate_ref == candidate_ref and record.plan_hash == plan_hash ): return record return None async def begin_validation_checkpoint( self, trace_id: str, candidate_ref: CandidateRef, *, plan: dict[str, Any], plan_hash: str, validated_at_sequence: int, material_usage_recorded: bool, ) -> CandidateValidationCheckpoint: trace, root = await self._require_bound_trace(trace_id) async with self._lock_for(root.trace_id): ledger = await self._load_ledger(root.trace_id) persisted = ledger.candidate(candidate_ref) if persisted != candidate_ref: raise CandidateStateError( "Candidate validation references an altered CandidateRef" ) self._validate_candidate_owner(persisted, trace, root) await self._assert_candidate_active(ledger, persisted) for raw in ledger.validation_checkpoints: checkpoint = CandidateValidationCheckpoint.model_validate(raw) if ( checkpoint.candidate_ref == candidate_ref and checkpoint.plan_hash == plan_hash ): return checkpoint checkpoint = CandidateValidationCheckpoint( candidate_ref=candidate_ref, plan_hash=plan_hash, validation_plan=plan, validated_at_sequence=validated_at_sequence, material_usage_recorded=material_usage_recorded, ) updated = ledger.model_copy(update={ "validation_checkpoints": ( *ledger.validation_checkpoints, checkpoint.model_dump(mode="json"), ), }) await self._persist_ledger(root.trace_id, updated) return checkpoint async def update_validation_checkpoint( self, trace_id: str, candidate_ref: CandidateRef, plan_hash: str, **updates: Any, ) -> CandidateValidationCheckpoint: mutable_fields = { "scope_results", "fixed_checks", "fixed_scope_errors", "quality_completed", "material_usage_recorded", "aggregate_result", } unknown_fields = set(updates) - mutable_fields if unknown_fields: raise CandidateStateError( "Candidate validation checkpoint fields are immutable: " f"{', '.join(sorted(unknown_fields))}" ) trace, root = await self._require_bound_trace(trace_id) async with self._lock_for(root.trace_id): ledger = await self._load_ledger(root.trace_id) persisted = ledger.candidate(candidate_ref) if persisted != candidate_ref: raise CandidateStateError( "Candidate validation references an altered CandidateRef" ) self._validate_candidate_owner(persisted, trace, root) checkpoints = [] result = None for raw in ledger.validation_checkpoints: checkpoint = CandidateValidationCheckpoint.model_validate(raw) if ( checkpoint.candidate_ref == candidate_ref and checkpoint.plan_hash == plan_hash ): checkpoint = CandidateValidationCheckpoint.model_validate({ **checkpoint.model_dump(mode="json"), **updates, }) result = checkpoint checkpoints.append(checkpoint.model_dump(mode="json")) if result is None: raise CandidateStateError("Candidate validation checkpoint not found") updated = ledger.model_copy(update={ "validation_checkpoints": tuple(checkpoints), }) await self._persist_ledger(root.trace_id, updated) return result async def record_validation( self, trace_id: str, record: CandidateValidationRecord, ) -> None: trace, root = await self._require_bound_trace(trace_id) async with self._lock_for(root.trace_id): ledger = await self._load_ledger(root.trace_id) persisted = ledger.candidate(record.candidate_ref) if persisted != record.candidate_ref: raise CandidateStateError( "Candidate validation references an altered CandidateRef" ) self._validate_candidate_owner(persisted, trace, root) await self._assert_candidate_active(ledger, persisted) retained = [] for raw in ledger.validations: item = CandidateValidationRecord.model_validate(raw) if ( item.candidate_ref == record.candidate_ref and item.plan_hash == record.plan_hash ): continue retained.append(raw) updated = ledger.model_copy(update={ "validations": ( *retained, record.model_dump(mode="json"), ), }) checkpoints = [] for raw in updated.validation_checkpoints: checkpoint = CandidateValidationCheckpoint.model_validate(raw) if ( checkpoint.candidate_ref == record.candidate_ref and checkpoint.plan_hash == record.plan_hash ): result = record.validation_result checkpoint = checkpoint.model_copy(update={ "scope_results": tuple(result.get("scope_results", [])), "aggregate_result": result, }) checkpoints.append(checkpoint.model_dump(mode="json")) updated = updated.model_copy(update={ "validation_checkpoints": tuple(checkpoints), }) await self._persist_ledger(root.trace_id, updated) if self.event_service is not None: await self.event_service.emit_after_commit( source_trace_id=record.candidate_ref.owner_trace_id, event_type="validation.completed", event_key=( "validation.completed:candidate:" f"{record.candidate_ref.candidate_id}:" f"{record.candidate_ref.revision}:{record.plan_hash}" ), effective_at_sequence=record.validated_at_sequence, payload={ "subject_type": "candidate", "candidate_ref": record.candidate_ref.model_dump( mode="json" ), "validation_result": record.validation_result, }, ) async def apply_review_actions( self, parent_trace_id: str, child_trace_id: str, *, report_refs: list[CandidateRef], actions: list[CandidateReviewAction], effective_at_sequence: int, command_id: str | None = None, ) -> list[CandidateLifecycleRecord]: """Apply parent-owned candidate decisions behind one root lock.""" parent = await self.store.get_trace(parent_trace_id) child = await self.store.get_trace(child_trace_id) if parent is None or child is None: raise CandidateStateError("Parent or child Trace not found") if child.parent_trace_id != parent_trace_id or child.uid != parent.uid: raise CandidateStateError("Candidate review requires a direct child") _parent, root = await self._require_bound_trace(parent_trace_id) child_root = child.context.get("root_trace_id") or child.trace_id if ( child_root != root.trace_id or child.context.get("application_ref") != self.binding.application_ref.model_dump(mode="json") ): raise CandidateStateError("Candidate review application/root mismatch") report_by_key = { (item.candidate_id, item.revision): item for item in report_refs } action_keys = [ (item.candidate_ref.candidate_id, item.candidate_ref.revision) for item in actions ] if len(action_keys) != len(set(action_keys)): raise CandidateStateError("Candidate review actions must be unique") async with self._lock_for(root.trace_id): ledger = await self._load_ledger(root.trace_id) batch_operation_id = self._review_batch_operation_id( parent_trace_id, child_trace_id, command_id=command_id, effective_at_sequence=effective_at_sequence, ) batch_input_hash = sha256(canonical_json([ item.model_dump(mode="json") for item in actions ]).encode("utf-8")).hexdigest() batch_operation = next( ( item for item in ledger.operations if item.operation_id == batch_operation_id ), None, ) operations_to_reserve: list[CandidateOperationRecord] = [] if batch_operation is not None: if batch_operation.input_hash != batch_input_hash: raise CandidateStateError( "Candidate review batch payload changed during recovery" ) else: batch_operation = CandidateOperationRecord( operation_id=batch_operation_id, command_id=command_id, operation="review_batch", input_hash=batch_input_hash, ) operations_to_reserve.append(batch_operation) existing_operation_ids = { item.operation_id for item in ledger.operations } for action_index, action in enumerate(actions): action_operation_id = self._review_operation_id( parent_trace_id, child_trace_id, action, command_id=command_id, effective_at_sequence=effective_at_sequence, action_index=action_index, ) if action_operation_id in existing_operation_ids: continue operations_to_reserve.append(CandidateOperationRecord( operation_id=action_operation_id, command_id=command_id, operation=action.action, input_hash=sha256( canonical_json(action).encode("utf-8") ).hexdigest(), candidate=CandidatePointer( candidate_id=action.candidate_ref.candidate_id, revision=action.candidate_ref.revision, ), )) if operations_to_reserve: if ( len(ledger.operations) + len(operations_to_reserve) > MAX_CANDIDATE_OPERATIONS_PER_ROOT ): raise CandidateStateError( "Candidate review operation limit exceeded before execution" ) ledger = ledger.model_copy(update={ "operations": ( *ledger.operations, *operations_to_reserve, ), }) # Persist the ordered batch envelope and every action intent in # one replacement. Besides freezing completeness, this is the # durable capacity reservation that keeps later commands from # stealing slots needed by a crash replay. await self._persist_ledger(root.trace_id, ledger) validation_by_key: dict[tuple[str, int], CandidateValidationRecord] = {} for raw in ledger.validations: record = CandidateValidationRecord.model_validate(raw) key = ( record.candidate_ref.candidate_id, record.candidate_ref.revision, ) validation_by_key[key] = record for action_index, (action, key) in enumerate(zip(actions, action_keys)): if key not in report_by_key or report_by_key[key] != action.candidate_ref: raise CandidateStateError( "Candidate action must reference the exact child TaskReport" ) persisted = ledger.candidate(action.candidate_ref) if persisted != action.candidate_ref or persisted.owner_trace_id != child_trace_id: raise CandidateStateError( "Candidate action references another Task or altered revision" ) await self._assert_candidate_active(ledger, persisted) record = validation_by_key.get(key) if record is None: raise CandidateStateError( "Candidate action requires framework validation" ) from cyber_agent.core.validation import ValidationResult outcome = ValidationResult.model_validate( record.validation_result ).outcome if action.action == "adopt" and outcome != "passed": raise CandidateStateError( "Only a passed candidate revision can be adopted" ) if action.action == "revise" and outcome == "passed": raise CandidateStateError( "Candidate revise requires a non-passed validation" ) state = await self._current_state(ledger, action.candidate_ref) expected_operation_id = self._review_operation_id( parent_trace_id, child_trace_id, action, command_id=command_id, effective_at_sequence=effective_at_sequence, action_index=action_index, ) same_completed_operation = any( item.operation_id == expected_operation_id and item.status == "completed" for item in ledger.operations ) if ( state not in {"proposed", "adoption_pending"} and not same_completed_operation ): raise CandidateStateError( f"Candidate already has terminal state: {state}" ) applied: list[CandidateLifecycleRecord] = [] for action_index, action in enumerate(actions): ledger, record = await self._apply_review_action( ledger, action, parent_trace_id=parent_trace_id, child_trace_id=child_trace_id, root_trace_id=root.trace_id, effective_at_sequence=effective_at_sequence, command_id=command_id, action_index=action_index, ) applied.append(record) operations = tuple( item.model_copy(update={"status": "completed", "error": None}) if item.operation_id == batch_operation_id else item for item in ledger.operations ) ledger = ledger.model_copy(update={"operations": operations}) await self._persist_ledger(root.trace_id, ledger) return applied async def _apply_review_action( self, ledger: CandidateLedger, action: CandidateReviewAction, *, parent_trace_id: str, child_trace_id: str, root_trace_id: str, effective_at_sequence: int, command_id: str | None, action_index: int, ) -> tuple[CandidateLedger, CandidateLifecycleRecord]: pointer = CandidatePointer( candidate_id=action.candidate_ref.candidate_id, revision=action.candidate_ref.revision, ) operation_id = self._review_operation_id( parent_trace_id, child_trace_id, action, command_id=command_id, effective_at_sequence=effective_at_sequence, action_index=action_index, ) input_hash = sha256( canonical_json(action).encode("utf-8") ).hexdigest() existing = next( (item for item in ledger.operations if item.operation_id == operation_id), None, ) if existing is not None: if existing.input_hash != input_hash: raise CandidateStateError("Candidate review payload changed during recovery") if existing.status == "completed": record = next( item for item in reversed(ledger.lifecycle) if item.operation_id == operation_id ) return ledger, record if existing.status == "failed": raise CandidateStateError(existing.error or "Candidate review failed") else: operation = CandidateOperationRecord( operation_id=operation_id, command_id=command_id, operation=action.action, input_hash=input_hash, candidate=pointer, ) ledger = ledger.model_copy(update={ "operations": (*ledger.operations, operation), }) if action.action != "adopt": record = CandidateLifecycleRecord( operation_id=operation_id, candidate=pointer, state="discarded", effective_at_sequence=effective_at_sequence, source_trace_id=parent_trace_id, reason=( f"retry_from=output: {action.reason}" if action.action == "revise" else action.reason ), ) ledger = self._finish_review_operation(ledger, operation_id, record) await self._persist_ledger(root_trace_id, ledger) await self._emit_lifecycle(record, root_trace_id) return ledger, record pending = CandidateLifecycleRecord( operation_id=operation_id, candidate=pointer, state="adoption_pending", effective_at_sequence=effective_at_sequence, source_trace_id=parent_trace_id, reason=action.reason, ) if ledger.current_state(pointer) != "adoption_pending": ledger = ledger.model_copy(update={ "lifecycle": (*ledger.lifecycle, pending), }) await self._persist_ledger(root_trace_id, ledger) await self._emit_lifecycle(pending, root_trace_id) try: receipt = CandidateAdoptionReceipt.model_validate( await self.repository.commit_adoption(CandidateAdoptionRequest( operation_id=operation_id, candidate_ref=action.candidate_ref, )) ) if receipt.operation_id != operation_id or not receipt.committed: raise CandidateStateError("Candidate adoption receipt is invalid") except Exception as exc: failed = CandidateLifecycleRecord( operation_id=operation_id, candidate=pointer, state="adoption_failed", effective_at_sequence=effective_at_sequence, source_trace_id=parent_trace_id, reason=str(exc)[:2_000], ) ledger = self._finish_review_operation( ledger, operation_id, failed, error=str(exc), ) await self._persist_ledger(root_trace_id, ledger) await self._emit_lifecycle(failed, root_trace_id) raise adopted = CandidateLifecycleRecord( operation_id=operation_id, candidate=pointer, state="adopted", effective_at_sequence=effective_at_sequence, source_trace_id=parent_trace_id, reason=action.reason, receipt=receipt, ) ledger = self._finish_review_operation(ledger, operation_id, adopted) # A failure here intentionally leaves the durable pending intent. The # Repository sees the same operation_id when the review is replayed. await self._persist_ledger(root_trace_id, ledger) await self._emit_lifecycle(adopted, root_trace_id) return ledger, adopted @staticmethod def _review_operation_id( parent_trace_id: str, child_trace_id: str, action: CandidateReviewAction, *, command_id: str | None = None, effective_at_sequence: int = 0, action_index: int = 0, ) -> str: del action command_hash = sha256( canonical_json({ "parent_trace_id": parent_trace_id, "child_trace_id": child_trace_id, "command_id": command_id or f"sequence:{effective_at_sequence}", "action_index": action_index, }).encode("utf-8") ).hexdigest() return f"candidate-review:{command_hash}" @staticmethod def _review_batch_operation_id( parent_trace_id: str, child_trace_id: str, *, command_id: str | None, effective_at_sequence: int, ) -> str: command_hash = sha256(canonical_json({ "parent_trace_id": parent_trace_id, "child_trace_id": child_trace_id, "command_id": command_id or f"sequence:{effective_at_sequence}", }).encode("utf-8")).hexdigest() return f"candidate-review-batch:{command_hash}" async def _emit_candidate_registered( self, candidate: CandidateRef, operation_id: str, ) -> None: if self.event_service is None: return await self.event_service.emit_after_commit( source_trace_id=candidate.owner_trace_id, event_type="candidate.version_registered", event_key=( f"candidate.version_registered:{candidate.candidate_id}:" f"{candidate.revision}" ), effective_at_sequence=candidate.created_at_sequence, payload={"candidate_ref": candidate.model_dump(mode="json")}, ) await self.event_service.emit_after_commit( source_trace_id=candidate.owner_trace_id, event_type="candidate.lifecycle_changed", event_key=( f"candidate.lifecycle_changed:{operation_id}:proposed" ), effective_at_sequence=candidate.created_at_sequence, payload={"lifecycle": CandidateLifecycleRecord( operation_id=operation_id, candidate=CandidatePointer( candidate_id=candidate.candidate_id, revision=candidate.revision, ), state="proposed", effective_at_sequence=candidate.created_at_sequence, source_trace_id=candidate.owner_trace_id, ).model_dump(mode="json")}, ) async def _emit_lifecycle( self, lifecycle: CandidateLifecycleRecord, root_trace_id: str, ) -> None: del root_trace_id if self.event_service is None or lifecycle.source_trace_id is None: return await self.event_service.emit_after_commit( source_trace_id=lifecycle.source_trace_id, event_type="candidate.lifecycle_changed", event_key=( f"candidate.lifecycle_changed:{lifecycle.operation_id}:" f"{lifecycle.state}" ), effective_at_sequence=lifecycle.effective_at_sequence, payload={"lifecycle": lifecycle.model_dump(mode="json")}, ) @staticmethod def _finish_review_operation( ledger: CandidateLedger, operation_id: str, lifecycle: CandidateLifecycleRecord, *, error: str | None = None, ) -> CandidateLedger: operations = tuple( item.model_copy(update={ "status": "failed" if error else "completed", "error": error[:2_000] if error else None, }) if item.operation_id == operation_id else item for item in ledger.operations ) return ledger.model_copy(update={ "operations": operations, "lifecycle": (*ledger.lifecycle, lifecycle), }) async def assert_rewind_allowed( self, trace_id: str, after_sequence: int, ) -> None: trace, root = await self._require_bound_trace(trace_id) async with self._lock_for(root.trace_id): ledger = await self._load_ledger(root.trace_id) for lifecycle in ledger.lifecycle: if lifecycle.state != "adopted": continue candidate = ledger.candidate(lifecycle.candidate) ancestor_ids: set[str] = set() for start_trace_id in { lifecycle.source_trace_id, candidate.owner_trace_id, }: current_trace_id = start_trace_id while current_trace_id and current_trace_id not in ancestor_ids: ancestor_ids.add(current_trace_id) current = await self.store.get_trace(current_trace_id) current_trace_id = ( current.parent_trace_id if current is not None else None ) crosses_review = ( lifecycle.source_trace_id == trace_id and lifecycle.effective_at_sequence > after_sequence ) crosses_version = ( candidate.owner_trace_id == trace_id and candidate.created_at_sequence > after_sequence ) incomparable_ancestor = ( trace_id in ancestor_ids and trace_id not in { lifecycle.source_trace_id, candidate.owner_trace_id, } ) if crosses_review or crosses_version or incomparable_ancestor: raise CandidateStateError( "Cannot rewind across a committed candidate adoption" ) @staticmethod def _candidate_operation( ledger: CandidateLedger, candidate: CandidateRef, ) -> CandidateOperationRecord: for operation in ledger.operations: if ( operation.operation in {"create", "fork", "merge"} and operation.candidate is not None and operation.candidate.candidate_id == candidate.candidate_id and operation.candidate.revision == candidate.revision ): return operation raise CandidateStateError( "Candidate has no persisted creation command" ) async def _sequence_is_active( self, trace_id: str, sequence: int, *, command_id: str | None, allow_untracked: bool, head_sequence: int | None = None, ) -> bool: trace = await self.store.get_trace(trace_id) if trace is None: return False all_messages = await self.store.get_trace_messages(trace_id) if not all_messages: # CandidateService has a direct internal port for tests and # controlled adapters. Production tool commands are always bound # to a persisted assistant tool_call_id and never take this path. return allow_untracked path = await self.store.get_main_path_messages( trace_id, trace.head_sequence if head_sequence is None else head_sequence, ) message = next((item for item in path if item.sequence == sequence), None) if message is None: return False if command_id is not None: return message.role == "tool" and message.tool_call_id == command_id return True async def _assert_candidate_active( self, ledger: CandidateLedger, candidate: CandidateRef, *, active_head_sequence: int | None = None, ) -> None: operation = self._candidate_operation(ledger, candidate) if not await self._sequence_is_active( candidate.owner_trace_id, candidate.created_at_sequence, command_id=operation.command_id, allow_untracked=operation.command_id is None, head_sequence=active_head_sequence, ): raise CandidateStateError( "Candidate is not on the current owner Trace branch" ) async def _current_state( self, ledger: CandidateLedger, pointer: CandidatePointer, *, active_trace_id: str | None = None, active_head_sequence: int | None = None, ) -> str | None: candidate = ledger.candidate(pointer) creation = self._candidate_operation(ledger, candidate) state = None operations = {item.operation_id: item for item in ledger.operations} for lifecycle in ledger.lifecycle: if ( lifecycle.candidate.candidate_id != pointer.candidate_id or lifecycle.candidate.revision != pointer.revision or lifecycle.source_trace_id is None ): continue operation = operations.get(lifecycle.operation_id) command_id = operation.command_id if operation is not None else None if await self._sequence_is_active( lifecycle.source_trace_id, lifecycle.effective_at_sequence, command_id=command_id, allow_untracked=creation.command_id is None, head_sequence=( active_head_sequence if lifecycle.source_trace_id == active_trace_id else None ), ): state = lifecycle.state return state async def _require_owner_trace(self, trace_id: str): trace, root = await self._require_bound_trace(trace_id) state = ensure_task_protocol(trace.context) if state.get("task_report") is not None: raise CandidateStateError("Candidates cannot change after TaskReport submission") if state.get("pending_reviews") or state.get("next_actions"): raise CandidateStateError("Resolve protocol lifecycle work before candidates") return trace, root async def _require_bound_trace(self, trace_id: str): trace = await self.store.get_trace(trace_id) if trace is None: raise CandidateStateError(f"Trace not found: {trace_id}") policy = require_mutable_trace_policy(trace.context) if not policy.requires_task_progress: raise CandidateStateError( "Candidates require Recursive revision 3" ) application_ref = trace.context.get("application_ref") if application_ref != self.binding.application_ref.model_dump(mode="json"): raise CandidateStateError("Candidate Trace application binding mismatch") root_trace_id = trace.context.get("root_trace_id") or trace.trace_id root = trace if trace.trace_id == root_trace_id else await self.store.get_trace( root_trace_id ) if root is None or root.uid != trace.uid: raise CandidateStateError("Candidate root or owner mismatch") if root.context.get("application_ref") != application_ref: raise CandidateStateError("Candidate root application mismatch") return trace, root async def _load_ledger(self, root_trace_id: str) -> CandidateLedger: raw = await self.store.get_candidate_ledger(root_trace_id) return CandidateLedger.model_validate(raw or {}) async def _persist_ledger( self, root_trace_id: str, ledger: CandidateLedger, ) -> None: """Validate the complete immutable aggregate before atomic replacement.""" validated = CandidateLedger.model_validate( ledger.model_dump(mode="json") ) await self.store.replace_candidate_ledger( root_trace_id, validated.model_dump(mode="json"), ) def _allocate_pointer( self, operation: str, trace_id: str, sequence: int, parents: list[CandidatePointer], ledger: CandidateLedger, ) -> CandidatePointer: if operation == "create": if parents: raise CandidateStateError("create does not accept parent_refs") digest = sha256(f"{trace_id}:{sequence}".encode("utf-8")).hexdigest() return CandidatePointer(candidate_id=f"cand_{digest[:24]}", revision=1) if operation == "fork": if len(parents) != 1: raise CandidateStateError("fork requires exactly one parent") parent = ledger.candidate(parents[0]) revisions = [ item.revision for item in ledger.candidates if item.candidate_id == parent.candidate_id ] return CandidatePointer( candidate_id=parent.candidate_id, revision=max(revisions) + 1, ) if operation == "merge": if len(parents) < 2: raise CandidateStateError("merge requires at least two parents") digest = sha256(f"{trace_id}:{sequence}:merge".encode("utf-8")).hexdigest() return CandidatePointer(candidate_id=f"cand_{digest[:24]}", revision=1) raise CandidateStateError(f"Unsupported candidate operation: {operation}") @staticmethod def _reserved_candidate_slots(ledger: CandidateLedger) -> int: """Count published revisions plus recoverable version intents. A pending create/fork/merge owns its eventual aggregate slot. Counting the reservation prevents a later command from consuming the final slot and making the earlier Repository side effect impossible to finalize. """ pending_versions = sum( item.status == "pending" and item.operation in {"create", "fork", "merge"} for item in ledger.operations ) return len(ledger.candidates) + pending_versions def _validate_parents(self, trace, root, parents: list[CandidateRef]) -> None: for parent in parents: self._validate_candidate_owner(parent, trace, root) def _validate_candidate_owner(self, candidate, trace, root) -> None: if ( candidate.application_ref != self.binding.application_ref or candidate.root_trace_id != root.trace_id or candidate.owner_trace_id != trace.trace_id ): raise CandidateStateError( "Candidate belongs to another application, root, or Task" ) @staticmethod def _complete_operation( ledger: CandidateLedger, operation_id: str, candidate: CandidateRef, ) -> CandidateLedger: operations = tuple( item.model_copy(update={"status": "completed", "error": None}) if item.operation_id == operation_id else item for item in ledger.operations ) lifecycle = (*ledger.lifecycle, CandidateLifecycleRecord( operation_id=operation_id, candidate=CandidatePointer( candidate_id=candidate.candidate_id, revision=candidate.revision, ), state="proposed", effective_at_sequence=candidate.created_at_sequence, source_trace_id=candidate.owner_trace_id, )) return CandidateLedger( candidates=(*ledger.candidates, candidate), lifecycle=lifecycle, operations=operations, validations=ledger.validations, validation_checkpoints=ledger.validation_checkpoints, ) @staticmethod def _fail_operation( ledger: CandidateLedger, operation_id: str, error: str, ) -> CandidateLedger: operations = tuple( item.model_copy(update={ "status": "failed", "error": error[:2_000], }) if item.operation_id == operation_id else item for item in ledger.operations ) return ledger.model_copy(update={"operations": operations})