diff --git a/docs/implementation-status.md b/docs/implementation-status.md index 8a5888b..1e1c7d2 100644 --- a/docs/implementation-status.md +++ b/docs/implementation-status.md @@ -1,7 +1,7 @@ # Implementation status Status: Alpha baseline -Last updated: 2026-08-20 +Last updated: 2026-08-21 This page records what the repository actually implements. The formal L2 documents describe the target architecture; an implemented vertical slice does not imply that every capability in its @@ -106,7 +106,7 @@ A4.0 conformance harness (delivered by PR #150): `generic_subprocess_runtime`; it does not perform runtime authority cutover. Issues #135/#136 remain open until the later conformance, chaos, parity, and cutover slices are complete. -A4.1a direct deterministic admission (this branch, CI-only): +A4.1a direct deterministic admission (CI-only): - The `managed_runtime_direct_cutover` gate is an explicit opt-in dependency of `managed_runtime_worker` and is absent from every default profile. Bootstrap rejects it unless @@ -118,10 +118,27 @@ A4.1a direct deterministic admission (this branch, CI-only): - Migration 0047 and the ORM check constraint reject an unbound managed admission while keeping legacy rows and explicit shadow admission compatible. Real PostgreSQL coverage verifies schema and repository round-trip behavior. -- This slice does not connect the Worker to managed authority, change - `ManagedRuntimeExecutionService`, perform finalization/outcome reconciliation, enable the gate - on the test server, or claim production cutover. A4.1b (execution/authority) and A4.2 - (reviewed/coordinated) remain open; issues #135/#136 remain open. +- This slice does not enable the gate on the test server or claim production cutover. + +A4.1b.1 managed DIRECT Worker authority (CI/test-only): + +- The Worker selects only the immutable `TaskRun.runtime_authority`. Managed Runs never call the + legacy runner and continue on the managed path when the new-Run admission gate is later disabled. +- Runtime prepare, fenced ownership, and the explicit `DISPATCHING` boundary use short + transactions; adapter/provider work remains outside a database transaction. Provider terminal + evidence and Task/Run/Attempt finalization, conservative budget settlement, quota release, + Inbox dedupe, and required Outbox events commit in one PostgreSQL UoW. +- `OUTCOME_UNKNOWN`/`LOST`, provider response loss, and an expired owner after the dispatch + boundary park the execution in explicit Runtime/Task/Run/Attempt reconciliation states without + redispatch. Stable fenced uncertainty evidence makes message replay idempotent. +- The slice accepts only mapping output with empty usage because pricing lineage is not yet part of + the managed observation path. Other successful provider results fail closed at the control-plane + result boundary. Real PostgreSQL tests cover atomic success/replay, injected rollback after + observation write, expired-owner parking, and stale-fence rollback. +- No reconciliation command or ordinary transition leaves `RECONCILIATION_REQUIRED`; the + evidence-driven privileged convergence path is A4.1b.2. The gate remains absent from defaults + and disabled on the server. A4.2 reviewed/coordinated and A4.3 production durability/rollout + remain open; issues #135/#136 remain open. ## Current runnable baseline diff --git a/docs/operations/runtime-direct-cutover-rollback.md b/docs/operations/runtime-direct-cutover-rollback.md index bd013d0..6bd774b 100644 --- a/docs/operations/runtime-direct-cutover-rollback.md +++ b/docs/operations/runtime-direct-cutover-rollback.md @@ -1,14 +1,14 @@ # Deterministic direct-runtime admission rollback -Last updated: 2026-08-20 +Last updated: 2026-08-21 -This runbook describes the A4.1a test-only admission gate. It is not a production runtime -cutover procedure. +This runbook describes the A4.1a/A4.1b.1 CI/test-only admission and managed DIRECT Worker path. +It is not a production runtime cutover procedure. ## Scope `managed_runtime_direct_cutover` is disabled in all default profiles and must remain disabled on -the test server until the later execution slice is explicitly qualified. When enabled, bootstrap +the test server. When enabled, bootstrap accepts it only in `test`/`testing` with the deterministic model provider and requires `managed_runtime_worker`. @@ -22,12 +22,16 @@ immutable persisted snapshot. `managed_runtime_direct_cutover=false`. 2. Restart the API with the normal profile/configuration. The body-less task-run API remains available and new Runs use legacy authority. -3. Do not rewrite existing `runtime_authority=managed` rows or change their Runtime Version. This - A4.1a slice has no Worker cutover, so those rows must not be claimed as evidence of a completed - managed execution path. -4. Inspect the persisted Run and outbox records before any later execution slice is enabled. - Reconciliation or data repair is a separate, explicitly approved operation; never silently - retry by creating a new ordinary Run. +3. Do not rewrite existing `runtime_authority=managed` rows or change their Runtime Version. + Disabling the admission gate affects new Runs only. Keep `managed_runtime_worker` and the + pinned built-in v2 adapter available until every existing managed Run is terminal or explicitly + parked for reconciliation; an existing managed Run must never fall back to legacy execution. +4. Inspect RuntimeExecution phase, current owner/fence, latest Attempt, Inbox, and reconciliation + Outbox evidence. `DISPATCHING` or later with an expired owner and no reattach proof must park as + `RECONCILIATION_REQUIRED`; it must not be redispatched or replaced by an ordinary Run. +5. A4.1b.1 deliberately has no exit from `RECONCILIATION_REQUIRED`. Escalate and preserve the + evidence until the privileged A4.1b.2 reconcile command is available. Manual status edits, + direct database repair, and blind provider retry are prohibited. Migration 0047 is expand-only and keeps legacy rows valid. A schema rollback is performed through the repository's tested Alembic downgrade window, not by manually dropping the managed admission diff --git a/docs/roadmap.md b/docs/roadmap.md index da85ced..fb964b4 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -169,12 +169,15 @@ Exit signal：用户可从模板创建公司、绑定真实 Agent，在不伪造 - [x] 非 LangGraph subprocess Agent 通过同一 conformance（A4.0，PR #150 已交付） - [x] A4.1a CI-only deterministic DIRECT 新 Run admission（gate、域模型、0047 持久化约束、 内置 LangGraph v2 校验） +- [x] A4.1b.1 CI/test-only managed DIRECT Worker authority（fenced dispatch、原子终结、 + reconciliation-required 停车；不含 reconcile command） - [ ] MCP write 和 fake external action 通过统一 Intent/Permit/Receipt/Reconciliation - [ ] Chaos smoke 证明核心 crash windows 收敛且无重复不可逆副作用 Exit signal：同一部署管理 LangGraph 与非 LangGraph Agent；两者使用同一 Task/Run/Attempt、 身份、治理、Artifact 和恢复语义，并由机器可读故障报告证明关键不变量。 -当前 A4.0 conformance harness 已在 PR #150 完成，A4.1a 已交付 gate/domain/persistence/admission -边界；两者均不满足完整 A4，也不关闭 #135/#136。A4.1b 仍需 Worker authority execution、 -finalization/outcome reconciliation，完整 A4 还需 chaos、parity 和 reviewed/coordinated cutover。 +当前 A4.0 conformance harness 已在 PR #150 完成，A4.1a admission 与 A4.1b.1 managed DIRECT +Worker authority/atomic parking 已交付。A4.1b.2 仍需受权限控制、证据驱动的 reconcile command； +完整 A4 还需 chaos、parity、reviewed/coordinated cutover 和生产 durable runtime。#135/#136 +继续保持开放。 diff --git a/src/agentmesh/application/managed_runtime_execution.py b/src/agentmesh/application/managed_runtime_execution.py index 71f648f..6c269bf 100644 --- a/src/agentmesh/application/managed_runtime_execution.py +++ b/src/agentmesh/application/managed_runtime_execution.py @@ -10,10 +10,13 @@ from __future__ import annotations from datetime import datetime, timezone -from uuid import UUID +from uuid import NAMESPACE_URL, UUID, uuid5 from agentmesh.application.ports import ( + ManagedRuntimeAuthoritativeResult, + ManagedRuntimeControlPlaneFailure, ManagedRuntimeExecutionPort, + ManagedRuntimePreDispatchFailure, RuntimeAssignmentBuilder, WorkflowWorkItem, ) @@ -22,7 +25,16 @@ from agentmesh.domain.errors import InvalidTaskTransition from agentmesh.domain.runtime_execution import RuntimeExecutionPhase from agentmesh.domain.tasks import AttemptStatus, Task, TaskAttempt, TaskRun -from agentmesh.runtime_sdk import ManagedAgentRuntime, RuntimeObservation +from agentmesh.runtime_sdk import ( + ErrorCategory, + ManagedAgentRuntime, + RetryDisposition, + RuntimeAssignment, + RuntimeError, + RuntimeObservation, + RuntimePhase, + canonical_digest, +) _PHASES = {phase.value: phase for phase in RuntimeExecutionPhase} @@ -150,12 +162,159 @@ def execute_shadow( evidence_id=observation.observation_id, ) + def execute_authoritative( + self, + task: Task, + run: TaskRun, + attempt: TaskAttempt, + *, + work_item: WorkflowWorkItem | None = None, + ) -> ManagedRuntimeAuthoritativeResult: + """Dispatch one managed-authoritative execution without committing its outcome.""" + execution_identity = run.runtime_execution_id or run.runtime_execution_intent_id + if run.runtime_authority != "managed" or run.runtime_version_id is None: + raise ValueError("Managed authority requires a pinned Runtime Version") + if execution_identity is None: + raise ValueError("Managed authority requires a Runtime execution intent") + now = datetime.now(timezone.utc) + if attempt.status is not AttemptStatus.RUNNING or _utc(attempt.lease_expires_at) <= now: + raise InvalidTaskTransition("Managed Runtime Attempt lease is not active") + try: + assignment = self._assignment_builder.assignment_for( + task, run, attempt, work_item=work_item + ) + report = self._adapter.validate(assignment) + if not report.valid: + raise ValueError("Managed Runtime assignment validation failed") + binder = getattr(self._adapter, "bind_context", None) + if binder is None: + raise ValueError("Managed Runtime adapter has no assignment backend") + binder(assignment, task, run, attempt, work_item) + except Exception as exc: + raise ManagedRuntimePreDispatchFailure( + "Managed Runtime assignment preparation failed" + ) from exc + expected_key = f"runtime-dispatch:{task.tenant_id}:{execution_identity}" + try: + execution = self._registry.prepare_execution( + run_id=run.id, + assignment_id=_uuid(assignment.assignment_id), + assignment_digest=assignment.assignment_digest or "", + dispatch_key=expected_key, + execution_id=execution_identity, + ) + except Exception as exc: + raise ManagedRuntimeControlPlaneFailure( + "Managed Runtime execution preparation did not commit" + ) from exc + if execution.phase is not RuntimeExecutionPhase.PREPARED: + return self._unknown_result( + execution.id, + assignment, + "runtime.reattach_unavailable", + observed_at=execution.updated_at, + dispatch_crossed=True, + ) + try: + execution = self._registry.claim_execution_owner( + execution_id=execution.id, + attempt_id=attempt.id, + fencing_token=attempt.fencing_token, + expected_owner_attempt_id=execution.current_owner_attempt_id, + expected_fencing_token=execution.current_fencing_token, + expected_version=execution.version, + claim_reason=( + "replacement" + if execution.current_owner_attempt_id is not None + else "initial" + ), + now=now, + ) + execution = self._registry.mark_execution_dispatching( + execution_id=execution.id, + attempt_id=attempt.id, + fencing_token=attempt.fencing_token, + ) + except Exception as exc: + raise ManagedRuntimeControlPlaneFailure( + "Managed Runtime dispatch boundary did not commit" + ) from exc + try: + receipt = self._adapter.dispatch(assignment, dispatch_key=expected_key) + observation = receipt.observation + if observation is None: + observation = self._adapter.inspect(receipt.handle) + self._validate_identity(execution.id, assignment, observation) + except Exception: + return self._unknown_result( + execution.id, + assignment, + "runtime.provider_outcome_unknown", + observed_at=execution.updated_at, + dispatch_crossed=True, + ) + return ManagedRuntimeAuthoritativeResult( + execution_id=execution.id, + assignment_id=_uuid(assignment.assignment_id), + assignment_digest=assignment.assignment_digest or "", + observation=observation, + dispatch_crossed=True, + ) + + @staticmethod + def _validate_identity( + execution_id: UUID, assignment: RuntimeAssignment, observation: object + ) -> None: + if type(observation) is not RuntimeObservation: + raise ValueError("Runtime observation type is inconsistent") + assignment_id = assignment.assignment_id + assignment_digest = assignment.assignment_digest + if ( + observation.runtime_execution_id != str(execution_id) + or observation.assignment_id != assignment_id + or observation.assignment_digest != assignment_digest + or not observation.phase.terminal + ): + raise ValueError("Runtime observation identity is inconsistent") + + @staticmethod + def _unknown_result( + execution_id: UUID, + assignment: RuntimeAssignment, + code: str, + *, + observed_at: datetime, + dispatch_crossed: bool, + ) -> ManagedRuntimeAuthoritativeResult: + assignment_id = assignment.assignment_id + assignment_digest = assignment.assignment_digest + observation = RuntimeObservation( + observation_id=str(uuid5(NAMESPACE_URL, f"{execution_id}:{code}")), + runtime_execution_id=str(execution_id), + assignment_id=assignment_id, + assignment_digest=assignment_digest, + phase=RuntimePhase.OUTCOME_UNKNOWN, + observed_at=_utc(observed_at), + provider_event_id=code, + error=RuntimeError( + code=code, + category=ErrorCategory.UNKNOWN, + message="Runtime provider outcome requires reconciliation", + retry_disposition=RetryDisposition.RECONCILE, + ), + ) + return ManagedRuntimeAuthoritativeResult( + execution_id=execution_id, + assignment_id=_uuid(assignment_id), + assignment_digest=assignment_digest, + observation=observation, + dispatch_crossed=dispatch_crossed, + ) + def _uuid(value: str): return UUID(value) def _observation_digest(observation: RuntimeObservation) -> str: - from agentmesh.runtime_sdk import canonical_digest - return canonical_digest(observation.to_dict()) diff --git a/src/agentmesh/application/ports.py b/src/agentmesh/application/ports.py index 319ba74..19f4e99 100644 --- a/src/agentmesh/application/ports.py +++ b/src/agentmesh/application/ports.py @@ -101,7 +101,7 @@ ToolExecutionAuthorization, ToolInvocation, ) -from agentmesh.runtime_sdk import RuntimeAssignment +from agentmesh.runtime_sdk import RuntimeAssignment, RuntimeObservation class TaskRepository(Protocol): @@ -1169,6 +1169,23 @@ def assignment_for( ) -> RuntimeAssignment: ... +@dataclass(frozen=True) +class ManagedRuntimeAuthoritativeResult: + execution_id: UUID + assignment_id: UUID + assignment_digest: str + observation: RuntimeObservation + dispatch_crossed: bool + + +class ManagedRuntimePreDispatchFailure(RuntimeError): + """Assignment was rejected before persistent execution/provider effects.""" + + +class ManagedRuntimeControlPlaneFailure(RuntimeError): + """A recoverable prepare/claim/dispatch-boundary transaction failed.""" + + class ManagedRuntimeExecutionPort(Protocol): """A transactional coordinator around the framework-neutral runtime port.""" @@ -1181,6 +1198,15 @@ def execute_shadow( work_item: WorkflowWorkItem | None = None, ) -> Any: ... + def execute_authoritative( + self, + task: Task, + run: TaskRun, + attempt: TaskAttempt, + *, + work_item: WorkflowWorkItem | None = None, + ) -> ManagedRuntimeAuthoritativeResult: ... + @dataclass(frozen=True) class AgentExecutionContext: diff --git a/src/agentmesh/application/runtime_services.py b/src/agentmesh/application/runtime_services.py index b66710e..882b1cd 100644 --- a/src/agentmesh/application/runtime_services.py +++ b/src/agentmesh/application/runtime_services.py @@ -339,6 +339,14 @@ def prepare_execution( uow.commit() return value + def get_execution_for_run(self, run_id: UUID) -> RuntimeExecution | None: + """Return the active or unresolved execution for recovery decisions.""" + self._require_enabled() + with self._uow_factory() as uow: + return uow.runtimes.get_active_or_unresolved_for_run( + run_id, tenant_id=self._tenant_id, for_update=False + ) + def prepare_execution_in_uow( self, uow: UnitOfWork, @@ -483,6 +491,45 @@ def record_observation( fencing_token: int | None = None, now: datetime | None = None, ) -> RuntimeObservationOutcome: + with self._uow_factory() as uow: + outcome = self.record_observation_in_uow( + uow, + execution_id=execution_id, + observation_id=observation_id, + observation_digest=observation_digest, + assignment_id=assignment_id, + assignment_digest=assignment_digest, + phase=phase, + provider_sequence=provider_sequence, + observed_at=observed_at, + evidence=evidence, + safe_summary=safe_summary, + attempt_id=attempt_id, + fencing_token=fencing_token, + now=now, + ) + uow.commit() + return outcome + + def record_observation_in_uow( + self, + uow: UnitOfWork, + *, + execution_id: UUID, + observation_id: str, + observation_digest: str, + assignment_id: UUID, + assignment_digest: str, + phase: RuntimeExecutionPhase, + provider_sequence: int | None, + observed_at: datetime, + evidence: dict[str, Any] | None = None, + safe_summary: str | None = None, + attempt_id: UUID | None = None, + fencing_token: int | None = None, + now: datetime | None = None, + ) -> RuntimeObservationOutcome: + """Record Runtime evidence in a caller-owned atomic transaction.""" self._require_enabled() timestamp = now or _now() evidence = {} if evidence is None else evidence @@ -509,51 +556,50 @@ def record_observation( safe_summary is not None and len(safe_summary) > 4096 ): raise InvalidTaskInput("Runtime observation evidence is invalid") - with self._uow_factory() as uow: - execution = uow.runtimes.get_execution( - execution_id, tenant_id=self._tenant_id, for_update=True - ) - if execution is None: - raise RuntimeExecutionNotFound("Runtime execution was not found") - prior = uow.runtimes.prior_observations( - execution_id, - tenant_id=self._tenant_id, - observation_id=observation_id, - digest=observation_digest, - ) - if ( - assignment_id != execution.assignment_id - or assignment_digest != execution.assignment_digest - ): - outcome = RuntimeObservationOutcome.CONFLICT - elif any( - item.observation_id == observation_id - and item.observation_digest != observation_digest - for item in prior - ): - outcome = RuntimeObservationOutcome.CONFLICT - elif prior or ( - provider_sequence is not None - and execution.provider_sequence is not None - and provider_sequence <= execution.provider_sequence - ): - outcome = RuntimeObservationOutcome.DUPLICATE - elif ( - execution.current_owner_attempt_id is None - or execution.current_fencing_token is None - or execution.current_owner_attempt_id != attempt_id - or execution.current_fencing_token != fencing_token - ): - outcome = RuntimeObservationOutcome.STALE_OWNER - elif ( - provider_sequence is not None - and execution.provider_sequence is not None - and provider_sequence > execution.provider_sequence + 1 - ): - outcome = RuntimeObservationOutcome.GAP - else: - outcome = RuntimeObservationOutcome.APPLIED - observation_record = RuntimeObservationEvidence( + execution = uow.runtimes.get_execution( + execution_id, tenant_id=self._tenant_id, for_update=True + ) + if execution is None: + raise RuntimeExecutionNotFound("Runtime execution was not found") + prior = uow.runtimes.prior_observations( + execution_id, + tenant_id=self._tenant_id, + observation_id=observation_id, + digest=observation_digest, + ) + if ( + assignment_id != execution.assignment_id + or assignment_digest != execution.assignment_digest + ): + outcome = RuntimeObservationOutcome.CONFLICT + elif any( + item.observation_id == observation_id + and item.observation_digest != observation_digest + for item in prior + ): + outcome = RuntimeObservationOutcome.CONFLICT + elif prior or ( + provider_sequence is not None + and execution.provider_sequence is not None + and provider_sequence <= execution.provider_sequence + ): + outcome = RuntimeObservationOutcome.DUPLICATE + elif ( + execution.current_owner_attempt_id is None + or execution.current_fencing_token is None + or execution.current_owner_attempt_id != attempt_id + or execution.current_fencing_token != fencing_token + ): + outcome = RuntimeObservationOutcome.STALE_OWNER + elif ( + provider_sequence is not None + and execution.provider_sequence is not None + and provider_sequence > execution.provider_sequence + 1 + ): + outcome = RuntimeObservationOutcome.GAP + else: + outcome = RuntimeObservationOutcome.APPLIED + observation_record = RuntimeObservationEvidence( id=uuid4(), tenant_id=self._tenant_id, runtime_execution_id=execution_id, @@ -569,22 +615,21 @@ def record_observation( processing_outcome=outcome, provider_event_present=False, evidence=evidence, - ) - uow.runtimes.add_observation(observation_record) - if outcome is RuntimeObservationOutcome.APPLIED: - try: - updated = execution.apply_observation( - phase=phase, provider_sequence=provider_sequence, now=timestamp - ) - except InvalidTaskTransition: - uow.runtimes.update_observation_outcome( - observation_record, outcome=RuntimeObservationOutcome.CONFLICT - ) - outcome = RuntimeObservationOutcome.CONFLICT - else: - uow.runtimes.save_execution(updated, tenant_id=self._tenant_id) - uow.commit() - return outcome + ) + uow.runtimes.add_observation(observation_record) + if outcome is RuntimeObservationOutcome.APPLIED: + try: + updated = execution.apply_observation( + phase=phase, provider_sequence=provider_sequence, now=timestamp + ) + except InvalidTaskTransition: + uow.runtimes.update_observation_outcome( + observation_record, outcome=RuntimeObservationOutcome.CONFLICT + ) + outcome = RuntimeObservationOutcome.CONFLICT + else: + uow.runtimes.save_execution(updated, tenant_id=self._tenant_id) + return outcome def claim_execution_owner( self, @@ -616,6 +661,36 @@ def claim_execution_owner( uow.commit() return updated + def mark_execution_dispatching( + self, + *, + execution_id: UUID, + attempt_id: UUID, + fencing_token: int, + now: datetime | None = None, + ) -> RuntimeExecution: + """Persist the provider side-effect boundary before dispatch.""" + self._require_enabled() + with self._uow_factory() as uow: + execution = uow.runtimes.get_execution( + execution_id, tenant_id=self._tenant_id, for_update=True + ) + if execution is None: + raise RuntimeExecutionNotFound("Runtime execution was not found") + if ( + execution.current_owner_attempt_id != attempt_id + or execution.current_fencing_token != fencing_token + ): + raise RuntimeExecutionConflict("Runtime execution owner is stale") + updated = execution.apply_observation( + phase=RuntimeExecutionPhase.DISPATCHING, + provider_sequence=execution.provider_sequence, + now=now or _now(), + ) + uow.runtimes.save_execution(updated, tenant_id=self._tenant_id) + uow.commit() + return updated + def request_lifecycle_operation( self, *, diff --git a/src/agentmesh/application/services.py b/src/agentmesh/application/services.py index cc9cd7c..ac4c5e6 100644 --- a/src/agentmesh/application/services.py +++ b/src/agentmesh/application/services.py @@ -7,13 +7,15 @@ from datetime import datetime, timedelta from hashlib import sha256 from typing import Any -from uuid import UUID, uuid4 +from uuid import NAMESPACE_URL, UUID, uuid4, uuid5 from agentmesh.application.budget_services import BudgetController from agentmesh.application.coordination_services import CoordinatedScheduler from agentmesh.application.memory_runtime_services import RuntimeMemoryService from agentmesh.application.ports import ( + ManagedRuntimeAuthoritativeResult, ManagedRuntimeExecutionPort, + ManagedRuntimePreDispatchFailure, UnitOfWorkFactory, WorkflowExecutionResult, WorkflowRunner, @@ -53,6 +55,10 @@ from agentmesh.domain.observability import UsageRecord from agentmesh.domain.planning import GoalContract from agentmesh.domain.registry import AgentVersion, AgentVersionStatus, normalize_agent_name +from agentmesh.domain.runtime_execution import ( + RuntimeExecutionPhase, + RuntimeObservationOutcome, +) from agentmesh.domain.tasks import ( AcceptanceCriterion, AttemptStatus, @@ -74,6 +80,7 @@ ToolExecutionAuthorization, ) from agentmesh.features import Feature, FeatureGateSet +from agentmesh.runtime_sdk import RuntimePhase, canonical_digest logger = logging.getLogger(__name__) @@ -715,6 +722,7 @@ def __init__( uow_factory: UnitOfWorkFactory, workflow_runner: WorkflowRunner, managed_execution_service: ManagedRuntimeExecutionPort | None = None, + runtime_registry_service: RuntimeRegistryService | None = None, worker_id: str, consumer_name: str, lease_duration: timedelta, @@ -729,6 +737,7 @@ def __init__( self._uow_factory = uow_factory self._workflow_runner = workflow_runner self._managed_execution_service = managed_execution_service + self._runtime_registry_service = runtime_registry_service self._worker_id = worker_id self._consumer_name = consumer_name self._lease_duration = lease_duration @@ -744,11 +753,76 @@ def __init__( def process(self, envelope: MessageEnvelope) -> bool: task_id, run_id = self._validate(envelope) + authority = self._persisted_runtime_authority( + envelope, task_id=task_id, run_id=run_id + ) + if authority == "managed" and ( + self._managed_execution_service is None + or self._runtime_registry_service is None + ): + raise InvalidTaskInput("Managed Runtime execution service is unavailable") + if authority == "managed": + assert self._runtime_registry_service is not None + execution = self._runtime_registry_service.get_execution_for_run(run_id) + if ( + execution is not None + and execution.phase is not RuntimeExecutionPhase.PREPARED + and not execution.phase.terminal + and self._park_expired_crossed_execution( + envelope, + task_id=task_id, + run_id=run_id, + execution=execution, + ) + ): + return True leased = self._acquire(envelope, task_id=task_id, run_id=run_id) if leased is None: return False task, run, attempt = leased + if run.runtime_authority == "managed": + assert self._managed_execution_service is not None + try: + with _AttemptLeaseRenewer( + service=self, + run_id=run.id, + attempt_id=attempt.id, + lease_token=attempt.lease_token, + interval=self._lease_renewal_interval, + ): + result = self._managed_execution_service.execute_authoritative( + task, run, attempt + ) + except ManagedRuntimePreDispatchFailure as exc: + self._finalize_failure( + envelope, + task_id, + run_id, + attempt.id, + f"Managed Runtime preparation failed: {type(exc).__name__}", + ) + return True + self._finalize_managed( + envelope, + task_id=task_id, + run_id=run_id, + attempt_id=attempt.id, + result=result, + ) + if self._research_materialization_service is not None: + try: + self._research_materialization_service.materialize_if_ready( + task_id, actor=self._worker_id + ) + except Exception: + logger.warning( + "Automatic research materialization failed for Task %s", + task_id, + exc_info=True, + ) + return True + renewer = _AttemptLeaseRenewer( service=self, run_id=run.id, @@ -827,6 +901,25 @@ def process(self, envelope: MessageEnvelope) -> bool: ) return True + def _persisted_runtime_authority( + self, + envelope: MessageEnvelope, + *, + task_id: UUID, + run_id: UUID, + ) -> str: + """Read immutable authority without consulting rollout gates or mutating state.""" + with self._uow_factory() as uow: + task = TaskApplicationService._get_task_or_raise(uow, task_id) + run = uow.runs.get(run_id) + if ( + run is None + or run.task_id != task.id + or task.tenant_id != envelope.tenant_id + ): + raise InvalidMessage("RunRequested references an unknown task run") + return run.runtime_authority + def _comparison_eligible(self, run: TaskRun) -> bool: """Return true only for an explicitly admitted, pinned A2 Run.""" return ( @@ -839,6 +932,235 @@ def _comparison_eligible(self, run: TaskRun) -> bool: and self._managed_execution_service is not None ) + def _park_expired_crossed_execution( + self, + envelope: MessageEnvelope, + *, + task_id: UUID, + run_id: UUID, + execution: Any, + ) -> bool: + """Fence an expired owner and atomically park a crossed dispatch boundary.""" + registry = self._runtime_registry_service + assert registry is not None + with self._uow_factory() as uow: + if uow.inbox.contains( + envelope.tenant_id, self._consumer_name, envelope.message_id + ): + return True + task = TaskApplicationService._get_task_or_raise(uow, task_id, for_update=True) + run = uow.runs.get(run_id, for_update=True) + latest = uow.attempts.latest_for_run(run_id, for_update=True) + if ( + run is None + or run.task_id != task.id + or run.runtime_authority != "managed" + or task.status is not TaskStatus.RUNNING + or run.status is not RunStatus.RUNNING + or latest is None + or latest.status is not AttemptStatus.RUNNING + or latest.lease_expires_at > utc_now() + or execution.current_owner_attempt_id != latest.id + or execution.current_fencing_token != latest.fencing_token + ): + return False + reason = "runtime.dispatch_outcome_unconfirmed" + observation_id = str( + uuid5(NAMESPACE_URL, f"{execution.id}:{reason}:{execution.version}") + ) + evidence = { + "kind": "control_plane_uncertainty", + "reason_code": reason, + "execution_phase": execution.phase.value, + } + outcome = registry.record_observation_in_uow( + uow, + execution_id=execution.id, + observation_id=observation_id, + observation_digest=canonical_digest( + { + "observation_id": observation_id, + "execution_id": str(execution.id), + "phase": RuntimeExecutionPhase.OUTCOME_UNKNOWN.value, + "observed_at": execution.updated_at, + "evidence": evidence, + } + ), + assignment_id=execution.assignment_id, + assignment_digest=execution.assignment_digest, + phase=RuntimeExecutionPhase.OUTCOME_UNKNOWN, + provider_sequence=None, + observed_at=execution.updated_at, + evidence=evidence, + safe_summary="Runtime dispatch outcome is unconfirmed", + attempt_id=latest.id, + fencing_token=latest.fencing_token, + ) + if outcome is not RuntimeObservationOutcome.APPLIED: + raise RunLeaseUnavailable( + f"Runtime recovery evidence cannot park from {outcome.value}" + ) + BudgetController.settle_attempt(task, latest, ()) + QuotaController.release_attempt(uow, latest) + task.require_runtime_reconciliation(run.id, reason) + run.require_runtime_reconciliation(reason) + latest.mark_outcome_unknown(reason) + uow.tasks.save(task) + uow.runs.save(run) + uow.attempts.save(latest) + uow.outbox.add( + self._runtime_reconciliation_event( + envelope, + task, + run, + latest, + execution_id=execution.id, + runtime_phase=RuntimeExecutionPhase.OUTCOME_UNKNOWN.value, + reason=reason, + ) + ) + uow.inbox.add(InboxMessage.processed(self._consumer_name, envelope)) + uow.commit() + return True + + def _finalize_managed( + self, + envelope: MessageEnvelope, + *, + task_id: UUID, + run_id: UUID, + attempt_id: UUID, + result: ManagedRuntimeAuthoritativeResult, + ) -> None: + registry = self._runtime_registry_service + if registry is None: + raise InvalidTaskInput("Managed Runtime Registry is unavailable") + if result.dispatch_crossed is not True: + raise InvalidTaskInput("Managed Runtime result lacks dispatch-boundary evidence") + observation = result.observation + phase = RuntimeExecutionPhase(observation.phase.value) + with self._uow_factory() as uow: + task, run, attempt = self._load_finalization_state( + uow, task_id, run_id, attempt_id + ) + if run.runtime_authority != "managed": + raise InvalidMessage("Managed finalization requires managed Run authority") + outcome = registry.record_observation_in_uow( + uow, + execution_id=result.execution_id, + observation_id=observation.observation_id, + observation_digest=canonical_digest(observation.to_dict()), + assignment_id=result.assignment_id, + assignment_digest=result.assignment_digest, + phase=phase, + provider_sequence=observation.provider_sequence, + observed_at=observation.observed_at, + evidence={ + "provider_event_id": observation.provider_event_id, + "snapshot_digest": observation.snapshot_digest, + "progress": dict(observation.progress), + }, + safe_summary="Managed Runtime authoritative observation", + attempt_id=attempt.id, + fencing_token=attempt.fencing_token, + ) + if outcome is not RuntimeObservationOutcome.APPLIED: + raise RunLeaseUnavailable( + f"Runtime observation cannot finalize from {outcome.value}" + ) + + budget_rejection = BudgetController.settle_attempt(task, attempt, ()) + QuotaController.release_attempt(uow, attempt) + if task.status is TaskStatus.CANCELED or run.status is RunStatus.CANCELED: + if attempt.status is AttemptStatus.RUNNING: + attempt.cancel() + elif observation.phase is RuntimePhase.SUCCEEDED: + if type(observation.output) is dict and not observation.usage: + output = dict(observation.output) + run.succeed(output) + attempt.succeed() + if budget_rejection is not None: + task.wait_for_budget(budget_rejection, candidate_output=output) + else: + task.complete(run.id, output) + else: + reason = "runtime.authoritative_result_rejected" + run.fail(reason) + attempt.fail(reason) + task.fail(run.id, reason) + elif observation.phase in { + RuntimePhase.OUTCOME_UNKNOWN, + RuntimePhase.LOST, + }: + reason = ( + observation.error.code + if observation.error is not None + else "runtime.reconciliation_required" + ) + task.require_runtime_reconciliation(run.id, reason) + run.require_runtime_reconciliation(reason) + attempt.mark_outcome_unknown(reason) + uow.outbox.add( + self._runtime_reconciliation_event( + envelope, + task, + run, + attempt, + execution_id=result.execution_id, + runtime_phase=observation.phase.value, + reason=reason, + ) + ) + else: + reason = ( + observation.error.code + if observation.error is not None + else f"runtime.{observation.phase.value.lower()}" + ) + run.fail(reason) + attempt.fail(reason) + task.fail(run.id, reason) + uow.tasks.save(task) + uow.runs.save(run) + uow.attempts.save(attempt) + uow.inbox.add(InboxMessage.processed(self._consumer_name, envelope)) + if ( + self._runtime_memory_service is not None + and task.status is TaskStatus.COMPLETED + ): + self._runtime_memory_service.capture_completed_task_in_unit_of_work( + uow, task + ) + uow.commit() + + @staticmethod + def _runtime_reconciliation_event( + envelope: MessageEnvelope, + task: Task, + run: TaskRun, + attempt: TaskAttempt, + *, + execution_id: UUID, + runtime_phase: str, + reason: str, + ) -> MessageEnvelope: + return MessageEnvelope.domain_event( + schema_name="agentmesh.runtime.reconciliation.required", + tenant_id=task.tenant_id, + aggregate_id=task.id, + causation_id=envelope.message_id, + producer="agentmesh-managed-runtime-worker-v1", + payload={ + "tenant_id": task.tenant_id, + "task_id": str(task.id), + "run_id": str(run.id), + "attempt_id": str(attempt.id), + "runtime_execution_id": str(execution_id), + "runtime_phase": runtime_phase, + "reason_code": reason, + }, + ) + def _record_runtime_shadow( self, envelope: MessageEnvelope, @@ -957,6 +1279,19 @@ def _acquire( if run.comparison_mode == "deterministic_shadow" and run.runtime_execution_id is None: raise InvalidMessage("Runtime comparison admission is incomplete") + if (task.status, run.status) == ( + TaskStatus.RECONCILIATION_REQUIRED, + RunStatus.RECONCILIATION_REQUIRED, + ): + uow.inbox.add(InboxMessage.processed(self._consumer_name, envelope)) + uow.commit() + return None + if ( + task.status is TaskStatus.RECONCILIATION_REQUIRED + or run.status is RunStatus.RECONCILIATION_REQUIRED + ): + raise InvalidMessage("RunRequested references inconsistent reconciliation state") + if run.status in {RunStatus.SUCCEEDED, RunStatus.FAILED, RunStatus.CANCELED}: uow.inbox.add(InboxMessage.processed(self._consumer_name, envelope)) uow.commit() diff --git a/src/agentmesh/bootstrap.py b/src/agentmesh/bootstrap.py index 8c7a301..eb64f0d 100644 --- a/src/agentmesh/bootstrap.py +++ b/src/agentmesh/bootstrap.py @@ -769,6 +769,7 @@ def transport_factory(api_key: str) -> OpenAIResponsesTransport: ) runtime_adapter = None managed_execution_service = None + worker_runtime_registry = None if feature_gates.is_enabled(Feature.MANAGED_RUNTIME_WORKER): if runtime_settings.environment.lower() not in {"test", "testing"}: raise InvalidFeatureConfiguration( @@ -780,12 +781,13 @@ def transport_factory(api_key: str) -> OpenAIResponsesTransport: state_store=EphemeralRuntimeStateStore(), lifecycle_controller=EphemeralRuntimeLifecycleController(), ) + worker_runtime_registry = RuntimeRegistryService( + uow_factory=uow_factory, + tenant_id=runtime_settings.tenant_id, + feature_gates=feature_gates, + ) managed_execution_service = ManagedRuntimeExecutionService( - registry=RuntimeRegistryService( - uow_factory=uow_factory, - tenant_id=runtime_settings.tenant_id, - feature_gates=feature_gates, - ), + registry=worker_runtime_registry, adapter=runtime_adapter, assignment_builder=runtime_adapter, ) @@ -823,6 +825,7 @@ def transport_factory(api_key: str) -> OpenAIResponsesTransport: uow_factory=uow_factory, workflow_runner=workflow_runner, managed_execution_service=managed_execution_service, + runtime_registry_service=worker_runtime_registry, worker_id=worker_id, consumer_name=runtime_settings.execution_consumer_name, lease_duration=timedelta(seconds=runtime_settings.run_lease_seconds), diff --git a/src/agentmesh/domain/runtime_execution.py b/src/agentmesh/domain/runtime_execution.py index b627d8a..5ebaa16 100644 --- a/src/agentmesh/domain/runtime_execution.py +++ b/src/agentmesh/domain/runtime_execution.py @@ -489,14 +489,21 @@ def claim( raise InvalidTaskTransition("Runtime execution owner changed before ownership claim") if self.phase.terminal: raise InvalidTaskTransition("Terminal Runtime execution cannot be claimed") - if self.current_owner_attempt_id is not None and ( - not replacement_authorized - or reattach_evidence is None - or reattach_evidence.execution_id != self.id - or reattach_evidence.assignment_digest != self.assignment_digest - or reattach_evidence.provider_execution_ref != self.provider_execution_ref - ): - raise InvalidTaskTransition("Active Runtime execution owner cannot be replaced") + if self.current_owner_attempt_id is not None: + safe_pre_dispatch_replacement = ( + self.phase is RuntimeExecutionPhase.PREPARED + and replacement_authorized + and reattach_evidence is None + ) + proven_reattach = ( + replacement_authorized + and reattach_evidence is not None + and reattach_evidence.execution_id == self.id + and reattach_evidence.assignment_digest == self.assignment_digest + and reattach_evidence.provider_execution_ref == self.provider_execution_ref + ) + if not safe_pre_dispatch_replacement and not proven_reattach: + raise InvalidTaskTransition("Active Runtime execution owner cannot be replaced") return replace( self, current_owner_attempt_id=attempt_id, diff --git a/src/agentmesh/domain/tasks.py b/src/agentmesh/domain/tasks.py index f010625..11eb3e6 100644 --- a/src/agentmesh/domain/tasks.py +++ b/src/agentmesh/domain/tasks.py @@ -25,6 +25,7 @@ class TaskStatus(str, Enum): REVIEWING = "REVIEWING" WAITING_APPROVAL = "WAITING_APPROVAL" WAITING_REMOTE = "WAITING_REMOTE" + RECONCILIATION_REQUIRED = "RECONCILIATION_REQUIRED" PAUSE_REQUESTED = "PAUSE_REQUESTED" PAUSED = "PAUSED" COMPLETED = "COMPLETED" @@ -38,6 +39,7 @@ class RunStatus(str, Enum): PAUSE_REQUESTED = "PAUSE_REQUESTED" PAUSED = "PAUSED" WAITING_REMOTE = "WAITING_REMOTE" + RECONCILIATION_REQUIRED = "RECONCILIATION_REQUIRED" SUCCEEDED = "SUCCEEDED" FAILED = "FAILED" CANCELED = "CANCELED" @@ -50,6 +52,7 @@ class AttemptStatus(str, Enum): FAILED = "FAILED" CANCELED = "CANCELED" LEASE_EXPIRED = "LEASE_EXPIRED" + OUTCOME_UNKNOWN = "OUTCOME_UNKNOWN" class TaskExecutionMode(str, Enum): @@ -373,6 +376,21 @@ def start(self, run_id: UUID) -> None: self.status = TaskStatus.RUNNING self._touch() + def require_runtime_reconciliation(self, run_id: UUID, reason: str) -> None: + self._require_active_run( + run_id, + "require Runtime reconciliation", + expected=TaskStatus.RUNNING, + ) + if self.execution_mode is not TaskExecutionMode.DIRECT: + raise InvalidTaskTransition( + "Only direct Tasks can require Runtime reconciliation" + ) + self.status = TaskStatus.RECONCILIATION_REQUIRED + self.output = None + self.error = _runtime_reconciliation_reason(reason) + self._touch() + def start_coordination(self) -> None: self._require_status(TaskStatus.CREATED, "start coordination") if self.execution_mode != TaskExecutionMode.COORDINATED: @@ -518,6 +536,10 @@ def fail(self, run_id: UUID, error: str) -> None: self._touch() def cancel(self) -> None: + if self.status is TaskStatus.RECONCILIATION_REQUIRED: + raise InvalidTaskTransition( + f"Cannot cancel task {self.id} while Runtime reconciliation is required" + ) if self.status in TERMINAL_TASK_STATUSES: raise InvalidTaskTransition( f"Cannot cancel task {self.id} from terminal status {self.status.value}" @@ -837,6 +859,16 @@ def start(self) -> None: if self.started_at is None: self.started_at = utc_now() + def require_runtime_reconciliation(self, reason: str) -> None: + if self.runtime_authority != "managed": + raise InvalidTaskTransition( + "Only managed Runs can require Runtime reconciliation" + ) + self._require_status(RunStatus.RUNNING, "require Runtime reconciliation") + self.status = RunStatus.RECONCILIATION_REQUIRED + self.output = None + self.error = _runtime_reconciliation_reason(reason) + def wait_for_remote(self) -> None: self._require_status(RunStatus.QUEUED, "wait for remote") self.status = RunStatus.WAITING_REMOTE @@ -1037,6 +1069,12 @@ def expire(self) -> None: self.status = AttemptStatus.LEASE_EXPIRED self.completed_at = utc_now() + def mark_outcome_unknown(self, reason: str) -> None: + self._require_running("mark outcome unknown") + self.status = AttemptStatus.OUTCOME_UNKNOWN + self.error = _runtime_reconciliation_reason(reason) + self.completed_at = utc_now() + def renew( self, *, @@ -1061,6 +1099,17 @@ def _require_running(self, action: str) -> None: ) +def _runtime_reconciliation_reason(reason: str) -> str: + normalized = reason.strip() if type(reason) is str else "" + if ( + not normalized + or len(normalized) > 512 + or any(ord(character) < 32 for character in normalized) + ): + raise InvalidTaskInput("Runtime reconciliation requires a bounded safe reason") + return normalized + + @dataclass(frozen=True) class TaskAggregate: task: Task diff --git a/src/agentmesh/infrastructure/runtime/langgraph_adapter.py b/src/agentmesh/infrastructure/runtime/langgraph_adapter.py index c116e92..d458124 100644 --- a/src/agentmesh/infrastructure/runtime/langgraph_adapter.py +++ b/src/agentmesh/infrastructure/runtime/langgraph_adapter.py @@ -478,6 +478,9 @@ def assignment_for( raise ValueError("Managed Runtime requires pinned Runtime and Agent Versions") input_value = work_item.input if work_item is not None else dict(task.input) objective = work_item.objective if work_item is not None else task.objective + runtime_execution_id = run.runtime_execution_id or run.runtime_execution_intent_id + if runtime_execution_id is None: + raise ValueError("Managed Runtime requires a Runtime execution identity") return RuntimeAssignment( assignment_id=str(uuid5(NAMESPACE_URL, f"agentmesh:assignment:{run.id}")), tenant_id=task.tenant_id, @@ -494,11 +497,14 @@ def assignment_for( objective=objective, structured_input=input_value, acceptance_contract={"criteria": [item.to_dict() for item in task.acceptance_criteria]}, - trace_context={"trace_id": attempt.trace_id}, + # Provider dispatch identity belongs to the immutable Run/Runtime + # intent. Attempt identity is control-plane fencing evidence and + # must not change the canonical assignment on replacement. + trace_context={"trace_id": f"runtime:{runtime_execution_id}"}, correlation_ids={ "task_id": str(task.id), "run_id": str(run.id), - "runtime_execution_id": str(run.runtime_execution_id or ""), + "runtime_execution_id": str(runtime_execution_id), }, ) diff --git a/tests/integration/test_managed_direct_worker_postgres.py b/tests/integration/test_managed_direct_worker_postgres.py new file mode 100644 index 0000000..a7ea8d1 --- /dev/null +++ b/tests/integration/test_managed_direct_worker_postgres.py @@ -0,0 +1,442 @@ +"""Real PostgreSQL evidence for A4.1b.1 managed DIRECT authority.""" + +from __future__ import annotations + +import os +from datetime import datetime, timedelta, timezone +from uuid import UUID, uuid4 + +import pytest +from sqlalchemy import create_engine, delete, func, select +from sqlalchemy.orm import Session, sessionmaker + +from agentmesh.application.managed_runtime_execution import ManagedRuntimeExecutionService +from agentmesh.application.quota_services import QuotaPolicyService +from agentmesh.application.runtime_services import RuntimeRegistryService +from agentmesh.application.services import RunExecutionService, TaskApplicationService +from agentmesh.bootstrap import seed_builtin_registry +from agentmesh.config import get_settings +from agentmesh.domain.budgets import BudgetSettlementSource, TaskBudget +from agentmesh.domain.errors import RunLeaseUnavailable +from agentmesh.domain.messaging import MessageEnvelope +from agentmesh.domain.quotas import QuotaScope +from agentmesh.domain.runtime_execution import ( + RuntimeObservationOutcome, +) +from agentmesh.domain.tasks import AttemptStatus, RunStatus, TaskStatus +from agentmesh.features import FeatureGateSet +from agentmesh.infrastructure.postgres.models import ( + InboxMessageRecord, + OutboxEventRecord, + QuotaReservationRecord, + RuntimeExecutionRecord, + RuntimeObservationRecord, +) +from agentmesh.infrastructure.postgres.uow import SqlAlchemyUnitOfWorkFactory +from agentmesh.infrastructure.runtime.langgraph_adapter import ( + EphemeralRuntimeLifecycleController, + EphemeralRuntimeStateStore, + LangGraphManagedAgentRuntime, +) +from agentmesh.runtime_sdk import RuntimeObservation, RuntimePhase + +pytestmark = [ + pytest.mark.postgres, + pytest.mark.skipif( + os.getenv("AGENTMESH_RUN_POSTGRES_TESTS") != "1", + reason="set AGENTMESH_RUN_POSTGRES_TESTS=1 to run service integration tests", + ), +] + + +class _PoisonLegacyRunner: + def run(self, *args, **kwargs): + raise AssertionError("managed authority must never call the legacy runner") + + +class _DeterministicBackend: + def __init__(self) -> None: + self.calls = 0 + + def bind(self, assignment, task, run, attempt, work_item) -> None: + return None + + def execute(self, assignment): + self.calls += 1 + return RuntimeObservation( + observation_id=str(uuid4()), + runtime_execution_id=assignment.correlation_ids["runtime_execution_id"], + assignment_id=assignment.assignment_id, + assignment_digest=assignment.assignment_digest, + phase=RuntimePhase.SUCCEEDED, + observed_at=datetime.now(timezone.utc), + provider_event_id="postgres-managed-success", + output={"managed": "postgres"}, + ) + + +class _FaultAfterEvidenceRegistry(RuntimeRegistryService): + def record_observation_in_uow(self, uow, **kwargs): + outcome = super().record_observation_in_uow(uow, **kwargs) + assert outcome is RuntimeObservationOutcome.APPLIED + raise RuntimeError("fault after Runtime evidence") + + +class _StaleParkingRegistry(RuntimeRegistryService): + def record_observation_in_uow(self, uow, **kwargs): + kwargs["attempt_id"] = uuid4() + return super().record_observation_in_uow(uow, **kwargs) + + +class _PoisonManagedExecution: + def __init__(self) -> None: + self.calls = 0 + + def execute_authoritative(self, *args, **kwargs): + self.calls += 1 + raise AssertionError("crossed execution must park without redispatch") + + +def _gates() -> FeatureGateSet: + return FeatureGateSet.from_config( + "full", + "managed_agent_runtime=true,managed_runtime_worker=true," + "managed_runtime_direct_cutover=true,identity_rbac=true,quota_admission=true", + ) + + +def _fixture(*, lease_duration=timedelta(minutes=5), registry_type=RuntimeRegistryService): + settings = get_settings() + engine = create_engine(settings.database_url) + factory = sessionmaker(bind=engine, expire_on_commit=False, class_=Session) + with factory() as session: + outbox_ids_before_seed = set(session.scalars(select(OutboxEventRecord.id))) + seed_builtin_registry(settings) + # Registry seeding emits durable domain events. Delete only the rows added + # by this fixture so shared or parallel-test events are never consumed. + with factory() as session: + outbox_ids_after_seed = set(session.scalars(select(OutboxEventRecord.id))) + seeded_outbox_ids = outbox_ids_after_seed - outbox_ids_before_seed + if seeded_outbox_ids: + session.execute( + delete(OutboxEventRecord).where( + OutboxEventRecord.id.in_(seeded_outbox_ids), + OutboxEventRecord.tenant_id == settings.tenant_id, + ) + ) + session.commit() + uow_factory = SqlAlchemyUnitOfWorkFactory(factory) + registry = registry_type( + uow_factory=uow_factory, + tenant_id=settings.tenant_id, + feature_gates=_gates(), + ) + tasks = TaskApplicationService( + uow_factory=uow_factory, + agent_id=settings.agent_id, + tenant_id=settings.tenant_id, + feature_gates=_gates(), + runtime_registry_service=registry, + ) + backend = _DeterministicBackend() + adapter = LangGraphManagedAgentRuntime( + backend=backend, + state_store=EphemeralRuntimeStateStore(), + lifecycle_controller=EphemeralRuntimeLifecycleController(), + ) + managed = ManagedRuntimeExecutionService( + registry=registry, + adapter=adapter, + assignment_builder=adapter, + ) + consumer = f"pg-managed-{uuid4().hex}" + worker = RunExecutionService( + uow_factory=uow_factory, + workflow_runner=_PoisonLegacyRunner(), + managed_execution_service=managed, + runtime_registry_service=registry, + worker_id=consumer, + consumer_name=consumer, + lease_duration=lease_duration, + feature_gates=_gates(), + ) + return engine, factory, registry, tasks, worker, backend, consumer, settings + + +def _request(tasks, tenant_id: str, factory, *, budget=None): + task_id = tasks.create_task( + f"postgres managed {uuid4().hex}", budget=budget + ).task.id + run = tasks.request_run(task_id).runs[0] + # These tests invoke the application worker directly. Remove the durable + # RunRequested row immediately so the later shared Redis relay vertical + # cannot publish a wakeup that this test already consumed out of band. + with factory() as session: + for record in session.scalars(select(OutboxEventRecord)): + payload = record.envelope.get("payload", {}) + if ( + record.envelope.get("schema_name") == "agentmesh.run.requested" + and str(payload.get("run_id", "")) == str(run.id) + ): + session.delete(record) + session.commit() + envelope = MessageEnvelope.run_requested( + tenant_id=tenant_id, task_id=task_id, run_id=run.id + ) + return task_id, run, envelope + + +def _cleanup_task_outbox(factory, task_id) -> None: + if task_id is None: + return + expected = str(task_id) + with factory() as session: + for record in session.scalars(select(OutboxEventRecord)): + envelope = record.envelope + payload = envelope.get("payload", {}) + if ( + str(payload.get("task_id", "")) == expected + or str(envelope.get("correlation_id", "")) == expected + ): + session.delete(record) + session.commit() + + +def test_postgres_managed_authoritative_success_is_atomic_and_replay_safe() -> None: + engine, factory, _registry, tasks, worker, backend, consumer, settings = _fixture() + task_id = None + try: + task_id, run, envelope = _request(tasks, settings.tenant_id, factory) + assert worker.process(envelope) is True + aggregate = tasks.get_task(task_id) + assert aggregate.task.status is TaskStatus.COMPLETED + assert aggregate.runs[0].status is RunStatus.SUCCEEDED + assert aggregate.attempts[0].status is AttemptStatus.SUCCEEDED + assert backend.calls == 1 + with factory() as session: + execution = session.scalar( + select(RuntimeExecutionRecord).where(RuntimeExecutionRecord.run_id == run.id) + ) + assert execution is not None and execution.phase == "SUCCEEDED" + assert session.scalar( + select(func.count()).select_from(RuntimeObservationRecord).where( + RuntimeObservationRecord.runtime_execution_id == execution.id, + RuntimeObservationRecord.processing_outcome == "APPLIED", + ) + ) == 1 + assert session.get( + InboxMessageRecord, + (settings.tenant_id, consumer, envelope.message_id), + ) is not None + assert worker.process(envelope) is False + assert backend.calls == 1 + with factory() as session: + assert session.scalar( + select(func.count()).select_from(RuntimeObservationRecord).join( + RuntimeExecutionRecord, + RuntimeExecutionRecord.id + == RuntimeObservationRecord.runtime_execution_id, + ).where(RuntimeExecutionRecord.run_id == run.id) + ) == 1 + finally: + _cleanup_task_outbox(factory, task_id) + engine.dispose() + + +def test_postgres_managed_finalization_fault_rolls_back_all_authority() -> None: + engine, factory, registry, tasks, worker, backend, _consumer, settings = _fixture() + task_id = None + try: + task_id, run, envelope = _request(tasks, settings.tenant_id, factory) + fault = _FaultAfterEvidenceRegistry( + uow_factory=SqlAlchemyUnitOfWorkFactory(factory), + tenant_id=settings.tenant_id, + feature_gates=_gates(), + ) + worker._runtime_registry_service = fault + with pytest.raises(RuntimeError, match="fault after Runtime evidence"): + worker.process(envelope) + aggregate = tasks.get_task(task_id) + assert aggregate.task.status is TaskStatus.RUNNING + assert aggregate.runs[0].status is RunStatus.RUNNING + assert aggregate.attempts[0].status is AttemptStatus.RUNNING + assert backend.calls == 1 + with factory() as session: + execution = session.scalar( + select(RuntimeExecutionRecord).where(RuntimeExecutionRecord.run_id == run.id) + ) + assert execution is not None and execution.phase == "DISPATCHING" + assert session.scalar( + select(func.count()).select_from(RuntimeObservationRecord).where( + RuntimeObservationRecord.runtime_execution_id == execution.id + ) + ) == 0 + assert session.scalar( + select(func.count()).select_from(InboxMessageRecord).where( + InboxMessageRecord.message_id == envelope.message_id + ) + ) == 0 + finally: + _cleanup_task_outbox(factory, task_id) + engine.dispose() + + +def test_postgres_expired_dispatching_owner_parks_atomically_once() -> None: + engine, factory, registry, tasks, worker, _backend, consumer, settings = _fixture( + lease_duration=timedelta(seconds=-1) + ) + task_id = None + try: + budget = TaskBudget.create(max_tokens=100, token_reservation_per_attempt=25) + task_id, run, envelope = _request( + tasks, settings.tenant_id, factory, budget=budget + ) + QuotaPolicyService( + SqlAlchemyUnitOfWorkFactory(factory), settings.tenant_id + ).put_policy( + scope=QuotaScope.TENANT, + project_id=None, + max_concurrent_attempts=1, + weight=1, + created_by="postgres-test", + ) + task, leased_run, attempt = worker._acquire( + envelope, task_id=task_id, run_id=run.id + ) + adapter = LangGraphManagedAgentRuntime( + backend=_DeterministicBackend(), + state_store=EphemeralRuntimeStateStore(), + lifecycle_controller=EphemeralRuntimeLifecycleController(), + ) + assignment = adapter.assignment_for(task, leased_run, attempt) + execution = registry.prepare_execution( + run_id=run.id, + assignment_id=UUID(assignment.assignment_id), + assignment_digest=assignment.assignment_digest, + execution_id=run.runtime_execution_intent_id, + ) + execution = registry.claim_execution_owner( + execution_id=execution.id, + attempt_id=attempt.id, + fencing_token=attempt.fencing_token, + expected_owner_attempt_id=None, + expected_fencing_token=None, + expected_version=execution.version, + now=datetime.now(timezone.utc) - timedelta(seconds=2), + ) + registry.mark_execution_dispatching( + execution_id=execution.id, + attempt_id=attempt.id, + fencing_token=attempt.fencing_token, + ) + poison = _PoisonManagedExecution() + worker._managed_execution_service = poison + assert worker.process(envelope) is True + parked = tasks.get_task(task_id) + assert parked.task.status is TaskStatus.RECONCILIATION_REQUIRED + assert parked.runs[0].status is RunStatus.RECONCILIATION_REQUIRED + assert parked.attempts[0].status is AttemptStatus.OUTCOME_UNKNOWN + assert ( + parked.attempts[0].budget_settlement_source + is BudgetSettlementSource.CONSERVATIVE_ESTIMATE + ) + assert poison.calls == 0 + with factory() as session: + execution_row = session.get(RuntimeExecutionRecord, execution.id) + assert execution_row is not None and execution_row.phase == "OUTCOME_UNKNOWN" + assert session.scalar( + select(func.count()).select_from(OutboxEventRecord).where( + OutboxEventRecord.envelope["schema_name"].astext + == "agentmesh.runtime.reconciliation.required", + OutboxEventRecord.envelope["payload"]["run_id"].astext == str(run.id), + ) + ) == 1 + assert session.get( + InboxMessageRecord, + (settings.tenant_id, consumer, envelope.message_id), + ) is not None + reservation = session.scalar( + select(QuotaReservationRecord).where( + QuotaReservationRecord.attempt_id == attempt.id + ) + ) + assert reservation is not None and reservation.released_at is not None + assert worker.process(envelope) is False + assert poison.calls == 0 + finally: + _cleanup_task_outbox(factory, task_id) + engine.dispose() + + +def test_postgres_stale_parking_evidence_rolls_back_domain_state() -> None: + engine, factory, registry, tasks, worker, _backend, _consumer, settings = _fixture( + lease_duration=timedelta(seconds=-1) + ) + task_id = None + try: + task_id, run, envelope = _request(tasks, settings.tenant_id, factory) + task, leased_run, attempt = worker._acquire( + envelope, task_id=task_id, run_id=run.id + ) + adapter = LangGraphManagedAgentRuntime( + backend=_DeterministicBackend(), + state_store=EphemeralRuntimeStateStore(), + lifecycle_controller=EphemeralRuntimeLifecycleController(), + ) + assignment = adapter.assignment_for(task, leased_run, attempt) + execution = registry.prepare_execution( + run_id=run.id, + assignment_id=UUID(assignment.assignment_id), + assignment_digest=assignment.assignment_digest, + execution_id=run.runtime_execution_intent_id, + ) + execution = registry.claim_execution_owner( + execution_id=execution.id, + attempt_id=attempt.id, + fencing_token=attempt.fencing_token, + expected_owner_attempt_id=None, + expected_fencing_token=None, + expected_version=execution.version, + now=datetime.now(timezone.utc) - timedelta(seconds=2), + ) + registry.mark_execution_dispatching( + execution_id=execution.id, + attempt_id=attempt.id, + fencing_token=attempt.fencing_token, + ) + stale = _StaleParkingRegistry( + uow_factory=SqlAlchemyUnitOfWorkFactory( + sessionmaker(bind=engine, expire_on_commit=False, class_=Session) + ), + tenant_id=settings.tenant_id, + feature_gates=_gates(), + ) + worker._runtime_registry_service = stale + with pytest.raises(RunLeaseUnavailable, match="STALE_OWNER"): + worker.process(envelope) + unchanged = tasks.get_task(task_id) + assert unchanged.task.status is TaskStatus.RUNNING + assert unchanged.runs[0].status is RunStatus.RUNNING + assert unchanged.attempts[0].status is AttemptStatus.RUNNING + with factory() as session: + assert session.scalar( + select(func.count()).select_from(RuntimeObservationRecord).where( + RuntimeObservationRecord.runtime_execution_id == execution.id + ) + ) == 0 + assert session.scalar( + select(func.count()).select_from(InboxMessageRecord).where( + InboxMessageRecord.message_id == envelope.message_id + ) + ) == 0 + assert session.scalar( + select(func.count()).select_from(OutboxEventRecord).where( + OutboxEventRecord.envelope["schema_name"].astext + == "agentmesh.runtime.reconciliation.required", + OutboxEventRecord.envelope["payload"]["run_id"].astext + == str(run.id), + ) + ) == 0 + finally: + _cleanup_task_outbox(factory, task_id) + engine.dispose() diff --git a/tests/test_domain_tasks.py b/tests/test_domain_tasks.py index e15b981..4cb5c7b 100644 --- a/tests/test_domain_tasks.py +++ b/tests/test_domain_tasks.py @@ -1,9 +1,21 @@ +from datetime import timedelta from uuid import uuid4 import pytest from agentmesh.domain.errors import InvalidTaskInput, InvalidTaskTransition, TaskExecutionFailed -from agentmesh.domain.tasks import RunStatus, Task, TaskRun, TaskStatus +from agentmesh.domain.tasks import ( + AcceptanceCriterion, + AcceptanceCriterionKind, + AttemptStatus, + RunStatus, + Task, + TaskAttempt, + TaskExecutionMode, + TaskRun, + TaskStatus, + utc_now, +) def test_task_execution_failure_preserves_task_identity() -> None: @@ -41,6 +53,79 @@ def test_task_happy_path() -> None: assert run.thread_id == str(run.id) +def test_managed_direct_execution_can_be_parked_for_runtime_reconciliation() -> None: + task = Task.create( + tenant_id="test", + objective="Reconcile an uncertain provider outcome", + execution_mode=TaskExecutionMode.DIRECT, + ) + run = TaskRun.request( + task.id, + "demo-agent", + runtime_version_id=uuid4(), + runtime_authority="managed", + ) + task.queue(run.id) + task.start(run.id) + run.start() + attempt = TaskAttempt.lease( + run_id=run.id, + worker_id="worker-a", + fencing_token=1, + lease_expires_at=utc_now() + timedelta(minutes=1), + ) + + task.require_runtime_reconciliation(run.id, "runtime.provider_outcome_unknown") + run.require_runtime_reconciliation("runtime.provider_outcome_unknown") + attempt.mark_outcome_unknown("runtime.provider_outcome_unknown") + + assert task.status is TaskStatus.RECONCILIATION_REQUIRED + assert task.output is None + assert run.status is RunStatus.RECONCILIATION_REQUIRED + assert run.completed_at is None + assert attempt.status is AttemptStatus.OUTCOME_UNKNOWN + assert attempt.completed_at is not None + with pytest.raises(InvalidTaskTransition): + task.cancel() + with pytest.raises(InvalidTaskTransition): + run.fail("ordinary failure") + with pytest.raises(InvalidTaskTransition): + attempt.succeed() + + +def test_runtime_reconciliation_state_is_fail_closed() -> None: + reviewed = Task.create( + tenant_id="test", + objective="Reviewed task", + execution_mode=TaskExecutionMode.REVIEWED, + acceptance_criteria=( + AcceptanceCriterion.create( + key="summary", + description="Summary exists", + kind=AcceptanceCriterionKind.OUTPUT_PATH_EXISTS, + path=("summary",), + ), + ), + max_revisions=1, + ) + legacy = TaskRun.request(reviewed.id, "demo-agent") + reviewed.queue(legacy.id) + reviewed.start(legacy.id) + legacy.start() + + with pytest.raises(InvalidTaskTransition): + reviewed.require_runtime_reconciliation(legacy.id, "runtime.lost") + with pytest.raises(InvalidTaskTransition): + legacy.require_runtime_reconciliation("runtime.lost") + with pytest.raises(InvalidTaskInput): + TaskAttempt.lease( + run_id=legacy.id, + worker_id="worker-a", + fencing_token=1, + lease_expires_at=utc_now() + timedelta(minutes=1), + ).mark_outcome_unknown("x" * 513) + + def test_completed_task_cannot_run_again() -> None: task = Task.create(tenant_id="test", objective="Complete once") run = TaskRun.request(task.id, "demo-agent") diff --git a/tests/test_managed_runtime_execution.py b/tests/test_managed_runtime_execution.py index eceea35..1c779bf 100644 --- a/tests/test_managed_runtime_execution.py +++ b/tests/test_managed_runtime_execution.py @@ -6,8 +6,9 @@ import pytest from agentmesh.application.managed_runtime_execution import ManagedRuntimeExecutionService +from agentmesh.application.ports import ManagedRuntimeControlPlaneFailure from agentmesh.domain.errors import InvalidTaskTransition -from agentmesh.domain.runtime_execution import RuntimeExecution +from agentmesh.domain.runtime_execution import RuntimeExecution, RuntimeExecutionPhase from agentmesh.domain.tasks import Task, TaskAttempt, TaskRun from agentmesh.infrastructure.runtime.langgraph_adapter import ( EphemeralRuntimeLifecycleController, @@ -79,6 +80,20 @@ def claim_execution_owner( def record_observation(self, **kwargs) -> None: self.observation_calls += 1 + def mark_execution_dispatching( + self, *, execution_id, attempt_id, fencing_token, now=None + ): + assert self.execution is not None + assert self.execution.id == execution_id + assert self.execution.current_owner_attempt_id == attempt_id + assert self.execution.current_fencing_token == fencing_token + self.execution = self.execution.apply_observation( + phase=RuntimeExecutionPhase.DISPATCHING, + provider_sequence=None, + now=now, + ) + return self.execution + class _BoundaryRegistry(_Registry): def __init__(self) -> None: @@ -104,6 +119,20 @@ def claim_execution_owner(self, **kwargs): self.active = False self.events.append("claim:end") + def mark_execution_dispatching(self, **kwargs): + self.events.append("mark:start") + self.active = True + try: + return super().mark_execution_dispatching(**kwargs) + finally: + self.active = False + self.events.append("mark:end") + + +class _MarkFailureRegistry(_Registry): + def mark_execution_dispatching(self, **kwargs): + raise RuntimeError("database unavailable") + class _BoundaryAdapter: def __init__(self, delegate: LangGraphManagedAgentRuntime, registry: _BoundaryRegistry) -> None: @@ -218,3 +247,92 @@ def test_adapter_calls_start_after_registry_prepare_and_claim_return() -> None: "dispatch", ] assert backend.execute_calls == 1 + + +def test_authoritative_execution_returns_uncommitted_observation() -> None: + service, task, run, attempt, backend, registry = _fixture() + run.runtime_authority = "managed" + + result = service.execute_authoritative(task, run, attempt) + + assert result.observation.phase is RuntimePhase.SUCCEEDED + assert result.observation.output == {"ok": True} + assert result.dispatch_crossed is True + assert backend.execute_calls == 1 + assert registry.observation_calls == 0 + assert registry.execution is not None + assert registry.execution.phase is RuntimeExecutionPhase.DISPATCHING + + +def test_authoritative_validation_precedes_persistent_execution_preparation() -> None: + _service, task, run, attempt, backend, _registry = _fixture() + run.runtime_authority = "managed" + registry = _BoundaryRegistry() + delegate = LangGraphManagedAgentRuntime( + backend=backend, + state_store=EphemeralRuntimeStateStore(), + lifecycle_controller=EphemeralRuntimeLifecycleController(), + ) + service = ManagedRuntimeExecutionService( + registry=registry, + adapter=_BoundaryAdapter(delegate, registry), + assignment_builder=delegate, + ) + + service.execute_authoritative(task, run, attempt) + + assert registry.events == [ + "validate", + "bind", + "prepare:start", + "prepare:end", + "claim:start", + "claim:end", + "mark:start", + "mark:end", + "dispatch", + ] + + +def test_replacement_attempt_keeps_canonical_assignment_identity() -> None: + _service, task, run, first, _backend, _registry = _fixture() + adapter = LangGraphManagedAgentRuntime( + backend=_CountingBackend(), + state_store=EphemeralRuntimeStateStore(), + lifecycle_controller=EphemeralRuntimeLifecycleController(), + ) + replacement = TaskAttempt.lease( + run_id=run.id, + worker_id="worker-b", + fencing_token=2, + lease_expires_at=datetime.now(timezone.utc) + timedelta(minutes=5), + ) + + first_assignment = adapter.assignment_for(task, run, first) + replacement_assignment = adapter.assignment_for(task, run, replacement) + + assert first_assignment.assignment_id == replacement_assignment.assignment_id + assert first_assignment.assignment_digest == replacement_assignment.assignment_digest + + +def test_mark_dispatching_failure_never_calls_provider_and_remains_prepared() -> None: + _service, task, run, attempt, backend, _registry = _fixture() + run.runtime_authority = "managed" + registry = _MarkFailureRegistry() + adapter = LangGraphManagedAgentRuntime( + backend=backend, + state_store=EphemeralRuntimeStateStore(), + lifecycle_controller=EphemeralRuntimeLifecycleController(), + ) + service = ManagedRuntimeExecutionService( + registry=registry, + adapter=adapter, + assignment_builder=adapter, + ) + + with pytest.raises(ManagedRuntimeControlPlaneFailure, match="did not commit"): + service.execute_authoritative(task, run, attempt) + + assert backend.execute_calls == 0 + assert registry.execution is not None + assert registry.execution.phase is RuntimeExecutionPhase.PREPARED diff --git a/tests/test_runtime_execution_domain.py b/tests/test_runtime_execution_domain.py index d687b8b..e2a64ae 100644 --- a/tests/test_runtime_execution_domain.py +++ b/tests/test_runtime_execution_domain.py @@ -114,6 +114,30 @@ def test_replacement_claim_requires_verified_reattach_evidence() -> None: assert updated.current_owner_attempt_id == replacement +def test_prepared_execution_allows_safe_replacement_without_reattach() -> None: + first = uuid4() + value = _execution().claim( + attempt_id=first, + fencing_token=1, + expected_owner_attempt_id=None, + expected_fencing_token=None, + expected_version=1, + ) + replacement = uuid4() + + updated = value.claim( + attempt_id=replacement, + fencing_token=2, + expected_owner_attempt_id=first, + expected_fencing_token=1, + expected_version=value.version, + replacement_authorized=True, + ) + + assert updated.phase is RuntimeExecutionPhase.PREPARED + assert updated.current_owner_attempt_id == replacement + + def test_exact_claim_replay_is_idempotent_before_stale_cas_checks() -> None: attempt_id = uuid4() value = _execution().claim( diff --git a/tests/test_task_service.py b/tests/test_task_service.py index 21c1c52..bacf8b4 100644 --- a/tests/test_task_service.py +++ b/tests/test_task_service.py @@ -1,13 +1,19 @@ import time -from datetime import timedelta +from dataclasses import replace +from datetime import datetime, timedelta, timezone from uuid import uuid4 import pytest from langgraph.checkpoint.memory import InMemorySaver +from agentmesh.application.ports import ( + ManagedRuntimeAuthoritativeResult, + ManagedRuntimeControlPlaneFailure, +) from agentmesh.application.registry_services import AgentRegistryService from agentmesh.application.runtime_comparison import RuntimeComparisonSnapshot from agentmesh.application.services import RunExecutionService, TaskApplicationService +from agentmesh.domain.budgets import TaskBudget from agentmesh.domain.coordination import CoordinatedPlan, SubtaskSpec from agentmesh.domain.errors import ( IdempotencyConflict, @@ -15,6 +21,11 @@ InvalidTaskTransition, RunLeaseUnavailable, ) +from agentmesh.domain.runtime_execution import ( + RuntimeExecution, + RuntimeExecutionPhase, + RuntimeObservationOutcome, +) from agentmesh.domain.tasks import ( AcceptanceCriterion, AcceptanceCriterionKind, @@ -23,6 +34,7 @@ Task, TaskExecutionMode, TaskStatus, + utc_now, ) from agentmesh.features import FeatureGateSet from agentmesh.orchestration.agent import ( @@ -30,6 +42,14 @@ DeterministicAgentExecutor, ) from agentmesh.orchestration.workflow import LangGraphWorkflowRunner +from agentmesh.runtime_sdk import ( + ErrorCategory, + RetryDisposition, + RuntimeError, + RuntimeObservation, + RuntimePhase, + canonical_digest, +) from tests.fakes import InMemoryUnitOfWorkFactory @@ -69,6 +89,90 @@ def require_builtin_langgraph_v2_in_uow(self, uow): return type("BuiltinVersion", (), {"id": self.version_id})() +class _PoisonWorkflowRunner: + def run(self, *args, **kwargs): + raise AssertionError("legacy WorkflowRunner must not execute a managed Run") + + +class _AuthoritativeManagedExecution: + def __init__(self, phase=RuntimePhase.SUCCEEDED, output=None, usage=None) -> None: + self.phase = phase + self.output = {"managed": True} if output is None else output + self.usage = {} if usage is None else usage + self.calls = 0 + + def execute_authoritative(self, task, run, attempt, **kwargs): + self.calls += 1 + execution_id = run.runtime_execution_id or run.runtime_execution_intent_id + assignment_id = uuid4() + digest = "a" * 64 + return ManagedRuntimeAuthoritativeResult( + execution_id=execution_id, + assignment_id=assignment_id, + assignment_digest=digest, + observation=RuntimeObservation( + observation_id=str(uuid4()), + runtime_execution_id=str(execution_id), + assignment_id=str(assignment_id), + assignment_digest=digest, + phase=self.phase, + observed_at=datetime.now(timezone.utc), + provider_event_id="managed-test", + output=self.output if self.phase is RuntimePhase.SUCCEEDED else None, + usage=self.usage, + error=( + RuntimeError( + code="runtime.provider_outcome_unknown", + category=ErrorCategory.UNKNOWN, + message="provider outcome unknown", + retry_disposition=RetryDisposition.RECONCILE, + ) + if self.phase in {RuntimePhase.OUTCOME_UNKNOWN, RuntimePhase.LOST} + else None + ), + ), + dispatch_crossed=True, + ) + + +class _ControlPlaneFailureManagedExecution: + def execute_authoritative(self, *args, **kwargs): + raise ManagedRuntimeControlPlaneFailure("claim conflict") + + +class _AtomicRuntimeRegistry: + def __init__(self, outcome=RuntimeObservationOutcome.APPLIED) -> None: + self.calls = 0 + self.execution = None + self.outcome = outcome + + def record_observation_in_uow(self, uow, **kwargs): + self.calls += 1 + return self.outcome + + def get_execution_for_run(self, run_id): + return self.execution + + +class _MemoryCaptureProbe: + def __init__(self) -> None: + self.captures = 0 + + def capture_completed_task_in_unit_of_work(self, uow, task): + self.captures += 1 + + +class _ResearchProbe: + def __init__(self, *, fail=False) -> None: + self.calls = 0 + self.fail = fail + + def materialize_if_ready(self, task_id, *, actor): + self.calls += 1 + if self.fail: + raise RuntimeError("research unavailable") + + def _execution_service_with_gates(uow_factory, gates, managed): workflow = LangGraphWorkflowRunner( agent_executor=DeterministicAgentExecutor(), @@ -145,6 +249,503 @@ def test_direct_cutover_admits_new_run_with_builtin_v2_and_stable_intent( assert runtime.calls == 1 +def test_worker_uses_persisted_managed_authority_and_never_legacy( + uow_factory: InMemoryUnitOfWorkFactory, + registry_service: AgentRegistryService, +) -> None: + admission = _BuiltinRuntimeAdmission() + tasks = TaskApplicationService( + uow_factory=uow_factory, + agent_id="test-agent", + tenant_id="test-tenant", + feature_gates=FeatureGateSet.from_config( + "full", + "managed_agent_runtime=true,managed_runtime_worker=true," + "managed_runtime_direct_cutover=true", + ), + runtime_registry_service=admission, + ) + task_id = tasks.create_task("managed authority").task.id + run = tasks.request_run(task_id).runs[0] + envelope = uow_factory.store.outbox[-1] + managed = _AuthoritativeManagedExecution() + registry = _AtomicRuntimeRegistry() + worker = RunExecutionService( + uow_factory=uow_factory, + workflow_runner=_PoisonWorkflowRunner(), + managed_execution_service=managed, + runtime_registry_service=registry, + worker_id="managed-worker", + consumer_name="managed-worker-v1", + lease_duration=timedelta(minutes=5), + # Rollout gates are deliberately off: persisted authority wins. + feature_gates=FeatureGateSet.from_config("minimal"), + ) + + assert run.runtime_authority == "managed" + assert worker.process(envelope) is True + completed = tasks.get_task(task_id) + assert completed.task.status is TaskStatus.COMPLETED + assert completed.task.output == {"managed": True} + assert completed.runs[0].status is RunStatus.SUCCEEDED + assert completed.attempts[0].status is AttemptStatus.SUCCEEDED + assert managed.calls == 1 + assert registry.calls == 1 + assert worker.process(envelope) is False + assert managed.calls == 1 + assert registry.calls == 1 + + +def test_managed_run_missing_service_fails_before_state_mutation( + uow_factory: InMemoryUnitOfWorkFactory, + registry_service: AgentRegistryService, +) -> None: + tasks = TaskApplicationService( + uow_factory=uow_factory, + agent_id="test-agent", + tenant_id="test-tenant", + feature_gates=FeatureGateSet.from_config( + "full", + "managed_agent_runtime=true,managed_runtime_worker=true," + "managed_runtime_direct_cutover=true", + ), + runtime_registry_service=_BuiltinRuntimeAdmission(), + ) + task_id = tasks.create_task("missing managed service").task.id + tasks.request_run(task_id) + envelope = uow_factory.store.outbox[-1] + worker = RunExecutionService( + uow_factory=uow_factory, + workflow_runner=_PoisonWorkflowRunner(), + worker_id="managed-worker", + consumer_name="managed-worker-v1", + lease_duration=timedelta(minutes=5), + ) + + with pytest.raises(InvalidTaskInput, match="unavailable"): + worker.process(envelope) + + unchanged = tasks.get_task(task_id) + assert unchanged.task.status is TaskStatus.READY + assert unchanged.runs[0].status is RunStatus.QUEUED + assert unchanged.attempts == [] + + +def test_managed_control_plane_failure_keeps_run_recoverable() -> None: + uow_factory = InMemoryUnitOfWorkFactory() + agents = AgentRegistryService(uow_factory=uow_factory, tenant_id="test-tenant") + agents.ensure_builtin_agent("test-agent") + tasks = TaskApplicationService( + uow_factory=uow_factory, + agent_id="test-agent", + tenant_id="test-tenant", + feature_gates=FeatureGateSet.from_config( + "full", + "managed_agent_runtime=true,managed_runtime_worker=true," + "managed_runtime_direct_cutover=true", + ), + runtime_registry_service=_BuiltinRuntimeAdmission(), + ) + task_id = tasks.create_task("recover control-plane failure").task.id + tasks.request_run(task_id) + envelope = uow_factory.store.outbox[-1] + worker = RunExecutionService( + uow_factory=uow_factory, + workflow_runner=_PoisonWorkflowRunner(), + managed_execution_service=_ControlPlaneFailureManagedExecution(), + runtime_registry_service=_AtomicRuntimeRegistry(), + worker_id="managed-worker", + consumer_name="managed-worker-v1", + lease_duration=timedelta(minutes=5), + ) + + with pytest.raises(ManagedRuntimeControlPlaneFailure): + worker.process(envelope) + + recoverable = tasks.get_task(task_id) + assert recoverable.task.status is TaskStatus.RUNNING + assert recoverable.runs[0].status is RunStatus.RUNNING + assert recoverable.attempts[0].status is AttemptStatus.RUNNING + assert not uow_factory.store.inbox + + +def test_unknown_managed_outcome_parks_once_without_redispatch( + uow_factory: InMemoryUnitOfWorkFactory, + registry_service: AgentRegistryService, +) -> None: + tasks = TaskApplicationService( + uow_factory=uow_factory, + agent_id="test-agent", + tenant_id="test-tenant", + feature_gates=FeatureGateSet.from_config( + "full", + "managed_agent_runtime=true,managed_runtime_worker=true," + "managed_runtime_direct_cutover=true", + ), + runtime_registry_service=_BuiltinRuntimeAdmission(), + ) + task_id = tasks.create_task("uncertain managed authority").task.id + tasks.request_run(task_id) + envelope = uow_factory.store.outbox[-1] + managed = _AuthoritativeManagedExecution(phase=RuntimePhase.OUTCOME_UNKNOWN) + memory = _MemoryCaptureProbe() + worker = RunExecutionService( + uow_factory=uow_factory, + workflow_runner=_PoisonWorkflowRunner(), + managed_execution_service=managed, + runtime_registry_service=_AtomicRuntimeRegistry(), + runtime_memory_service=memory, + worker_id="managed-worker", + consumer_name="managed-worker-v1", + lease_duration=timedelta(minutes=5), + ) + + assert worker.process(envelope) is True + parked = tasks.get_task(task_id) + assert parked.task.status is TaskStatus.RECONCILIATION_REQUIRED + assert parked.runs[0].status is RunStatus.RECONCILIATION_REQUIRED + assert parked.attempts[0].status is AttemptStatus.OUTCOME_UNKNOWN + events = [ + item + for item in uow_factory.store.outbox + if item.schema_name == "agentmesh.runtime.reconciliation.required" + ] + assert len(events) == 1 + assert events[0].payload["reason_code"] == "runtime.provider_outcome_unknown" + assert worker.process(envelope) is False + assert managed.calls == 1 + assert memory.captures == 0 + assert len( + [ + item + for item in uow_factory.store.outbox + if item.schema_name == "agentmesh.runtime.reconciliation.required" + ] + ) == 1 + + +def test_expired_dispatching_owner_parks_before_replacement_attempt() -> None: + uow_factory = InMemoryUnitOfWorkFactory() + agents = AgentRegistryService(uow_factory=uow_factory, tenant_id="test-tenant") + agents.ensure_builtin_agent("test-agent") + tasks = TaskApplicationService( + uow_factory=uow_factory, + agent_id="test-agent", + tenant_id="test-tenant", + feature_gates=FeatureGateSet.from_config( + "full", + "managed_agent_runtime=true,managed_runtime_worker=true," + "managed_runtime_direct_cutover=true", + ), + runtime_registry_service=_BuiltinRuntimeAdmission(), + ) + task_id = tasks.create_task("recover crossed dispatch").task.id + run = tasks.request_run(task_id).runs[0] + envelope = uow_factory.store.outbox[-1] + registry = _AtomicRuntimeRegistry() + managed = _AuthoritativeManagedExecution() + worker = RunExecutionService( + uow_factory=uow_factory, + workflow_runner=_PoisonWorkflowRunner(), + managed_execution_service=managed, + runtime_registry_service=registry, + worker_id="managed-worker", + consumer_name="managed-worker-v1", + lease_duration=timedelta(seconds=-1), + ) + _task, _run, expired_owner = worker._acquire( + envelope, task_id=task_id, run_id=run.id + ) + execution_id = run.runtime_execution_intent_id + execution = RuntimeExecution.prepare( + tenant_id="test-tenant", + run_id=run.id, + runtime_version_id=run.runtime_version_id, + assignment_id=uuid4(), + assignment_digest="a" * 64, + dispatch_key=f"runtime-dispatch:test-tenant:{execution_id}", + dispatch_digest=canonical_digest({"execution": str(execution_id)}), + execution_id=execution_id, + ).claim( + attempt_id=expired_owner.id, + fencing_token=expired_owner.fencing_token, + expected_owner_attempt_id=None, + expected_fencing_token=None, + expected_version=1, + ) + registry.execution = execution.apply_observation( + phase=RuntimeExecutionPhase.DISPATCHING, + provider_sequence=None, + ) + + assert worker.process(envelope) is True + parked = tasks.get_task(task_id) + assert [attempt.status for attempt in parked.attempts] == [AttemptStatus.OUTCOME_UNKNOWN] + assert parked.task.status is TaskStatus.RECONCILIATION_REQUIRED + assert parked.runs[0].status is RunStatus.RECONCILIATION_REQUIRED + assert managed.calls == 0 + assert registry.calls == 1 + + +def test_stale_crossed_execution_parking_rolls_back_task_and_attempt() -> None: + uow_factory = InMemoryUnitOfWorkFactory() + agents = AgentRegistryService(uow_factory=uow_factory, tenant_id="test-tenant") + agents.ensure_builtin_agent("test-agent") + tasks = TaskApplicationService( + uow_factory=uow_factory, + agent_id="test-agent", + tenant_id="test-tenant", + feature_gates=FeatureGateSet.from_config( + "full", + "managed_agent_runtime=true,managed_runtime_worker=true," + "managed_runtime_direct_cutover=true", + ), + runtime_registry_service=_BuiltinRuntimeAdmission(), + ) + task_id = tasks.create_task("stale crossed dispatch").task.id + run = tasks.request_run(task_id).runs[0] + envelope = uow_factory.store.outbox[-1] + registry = _AtomicRuntimeRegistry(RuntimeObservationOutcome.STALE_OWNER) + worker = RunExecutionService( + uow_factory=uow_factory, + workflow_runner=_PoisonWorkflowRunner(), + managed_execution_service=_AuthoritativeManagedExecution(), + runtime_registry_service=registry, + worker_id="managed-worker", + consumer_name="managed-worker-v1", + lease_duration=timedelta(seconds=-1), + ) + _task, _run, owner = worker._acquire( + envelope, task_id=task_id, run_id=run.id + ) + execution_id = run.runtime_execution_intent_id + execution = RuntimeExecution.prepare( + tenant_id="test-tenant", + run_id=run.id, + runtime_version_id=run.runtime_version_id, + assignment_id=uuid4(), + assignment_digest="a" * 64, + dispatch_key=f"runtime-dispatch:test-tenant:{execution_id}", + dispatch_digest=canonical_digest({"execution": str(execution_id)}), + execution_id=execution_id, + ).claim( + attempt_id=owner.id, + fencing_token=owner.fencing_token, + expected_owner_attempt_id=None, + expected_fencing_token=None, + expected_version=1, + ) + registry.execution = execution.apply_observation( + phase=RuntimeExecutionPhase.DISPATCHING, + provider_sequence=None, + ) + + with pytest.raises(RunLeaseUnavailable, match="STALE_OWNER"): + worker.process(envelope) + + unchanged = tasks.get_task(task_id) + assert unchanged.task.status is TaskStatus.RUNNING + assert unchanged.runs[0].status is RunStatus.RUNNING + assert unchanged.attempts[0].status is AttemptStatus.RUNNING + assert not uow_factory.store.inbox + assert not [ + item + for item in uow_factory.store.outbox + if item.schema_name == "agentmesh.runtime.reconciliation.required" + ] + + +def test_managed_success_with_usage_fails_control_plane_result() -> None: + uow_factory = InMemoryUnitOfWorkFactory() + registry = AgentRegistryService(uow_factory=uow_factory, tenant_id="test-tenant") + registry.ensure_builtin_agent("test-agent") + admission = _BuiltinRuntimeAdmission() + tasks = TaskApplicationService( + uow_factory=uow_factory, + agent_id="test-agent", + tenant_id="test-tenant", + feature_gates=FeatureGateSet.from_config( + "full", + "managed_agent_runtime=true,managed_runtime_worker=true," + "managed_runtime_direct_cutover=true", + ), + runtime_registry_service=admission, + ) + task_id = tasks.create_task("reject unpriced usage").task.id + tasks.request_run(task_id) + envelope = uow_factory.store.outbox[-1] + worker = RunExecutionService( + uow_factory=uow_factory, + workflow_runner=_PoisonWorkflowRunner(), + managed_execution_service=_AuthoritativeManagedExecution(usage={"total": 1}), + runtime_registry_service=_AtomicRuntimeRegistry(), + worker_id="managed-worker", + consumer_name="managed-worker-v1", + lease_duration=timedelta(minutes=5), + ) + + assert worker.process(envelope) is True + rejected = tasks.get_task(task_id) + assert rejected.task.status is TaskStatus.FAILED + assert rejected.task.error == "runtime.authoritative_result_rejected" + assert rejected.runs[0].status is RunStatus.FAILED + assert rejected.attempts[0].status is AttemptStatus.FAILED + + +def test_late_managed_success_does_not_overwrite_cancellation() -> None: + uow_factory = InMemoryUnitOfWorkFactory() + agents = AgentRegistryService(uow_factory=uow_factory, tenant_id="test-tenant") + agents.ensure_builtin_agent("test-agent") + tasks = TaskApplicationService( + uow_factory=uow_factory, + agent_id="test-agent", + tenant_id="test-tenant", + feature_gates=FeatureGateSet.from_config( + "full", + "managed_agent_runtime=true,managed_runtime_worker=true," + "managed_runtime_direct_cutover=true", + ), + runtime_registry_service=_BuiltinRuntimeAdmission(), + ) + task_id = tasks.create_task("cancel during managed dispatch").task.id + run = tasks.request_run(task_id).runs[0] + envelope = uow_factory.store.outbox[-1] + registry = _AtomicRuntimeRegistry() + worker = RunExecutionService( + uow_factory=uow_factory, + workflow_runner=_PoisonWorkflowRunner(), + managed_execution_service=_AuthoritativeManagedExecution(), + runtime_registry_service=registry, + worker_id="managed-worker", + consumer_name="managed-worker-v1", + lease_duration=timedelta(minutes=5), + ) + task, leased_run, attempt = worker._acquire( + envelope, task_id=task_id, run_id=run.id + ) + result = _AuthoritativeManagedExecution().execute_authoritative( + task, leased_run, attempt + ) + tasks.cancel_task(task_id) + + worker._finalize_managed( + envelope, + task_id=task_id, + run_id=run.id, + attempt_id=attempt.id, + result=result, + ) + + canceled = tasks.get_task(task_id) + assert canceled.task.status is TaskStatus.CANCELED + assert canceled.runs[0].status is RunStatus.CANCELED + assert canceled.attempts[0].status is AttemptStatus.CANCELED + assert canceled.task.output is None + assert registry.calls == 1 + + +def test_managed_success_honors_budget_deadline_during_atomic_finalization() -> None: + uow_factory = InMemoryUnitOfWorkFactory() + agents = AgentRegistryService(uow_factory=uow_factory, tenant_id="test-tenant") + agents.ensure_builtin_agent("test-agent") + tasks = TaskApplicationService( + uow_factory=uow_factory, + agent_id="test-agent", + tenant_id="test-tenant", + feature_gates=FeatureGateSet.from_config( + "full", + "managed_agent_runtime=true,managed_runtime_worker=true," + "managed_runtime_direct_cutover=true", + ), + runtime_registry_service=_BuiltinRuntimeAdmission(), + ) + task_id = tasks.create_task( + "managed budget deadline", + budget=TaskBudget.create( + deadline=utc_now() + timedelta(minutes=5), + max_tokens=100, + token_reservation_per_attempt=10, + ), + ).task.id + run = tasks.request_run(task_id).runs[0] + envelope = uow_factory.store.outbox[-1] + worker = RunExecutionService( + uow_factory=uow_factory, + workflow_runner=_PoisonWorkflowRunner(), + managed_execution_service=_AuthoritativeManagedExecution(), + runtime_registry_service=_AtomicRuntimeRegistry(), + worker_id="managed-worker", + consumer_name="managed-worker-v1", + lease_duration=timedelta(minutes=5), + ) + task, leased_run, attempt = worker._acquire( + envelope, task_id=task_id, run_id=run.id + ) + result = _AuthoritativeManagedExecution().execute_authoritative( + task, leased_run, attempt + ) + with uow_factory() as uow: + current = uow.tasks.get(task_id, for_update=True) + assert current is not None and current.budget is not None + current.budget = replace( + current.budget, deadline=utc_now() - timedelta(seconds=1) + ) + uow.tasks.save(current) + uow.commit() + + worker._finalize_managed( + envelope, + task_id=task_id, + run_id=run.id, + attempt_id=attempt.id, + result=result, + ) + + waiting = tasks.get_task(task_id) + assert waiting.task.status is TaskStatus.WAITING_APPROVAL + assert waiting.task.error == "budget_deadline_exceeded" + assert waiting.task.candidate_output == {"managed": True} + assert waiting.runs[0].status is RunStatus.SUCCEEDED + assert waiting.attempts[0].status is AttemptStatus.SUCCEEDED + + +def test_managed_completion_captures_memory_and_research_failure_is_non_authoritative() -> None: + uow_factory = InMemoryUnitOfWorkFactory() + agents = AgentRegistryService(uow_factory=uow_factory, tenant_id="test-tenant") + agents.ensure_builtin_agent("test-agent") + tasks = TaskApplicationService( + uow_factory=uow_factory, + agent_id="test-agent", + tenant_id="test-tenant", + feature_gates=FeatureGateSet.from_config( + "full", + "managed_agent_runtime=true,managed_runtime_worker=true," + "managed_runtime_direct_cutover=true", + ), + runtime_registry_service=_BuiltinRuntimeAdmission(), + ) + task_id = tasks.create_task("managed memory parity").task.id + tasks.request_run(task_id) + memory = _MemoryCaptureProbe() + research = _ResearchProbe(fail=True) + worker = RunExecutionService( + uow_factory=uow_factory, + workflow_runner=_PoisonWorkflowRunner(), + managed_execution_service=_AuthoritativeManagedExecution(), + runtime_registry_service=_AtomicRuntimeRegistry(), + runtime_memory_service=memory, + research_materialization_service=research, + worker_id="managed-worker", + consumer_name="managed-worker-v1", + lease_duration=timedelta(minutes=5), + ) + + assert worker.process(uow_factory.store.outbox[-1]) is True + assert tasks.get_task(task_id).task.status is TaskStatus.COMPLETED + assert memory.captures == 1 + assert research.calls == 1 + + def test_direct_cutover_gate_off_keeps_new_runs_legacy_and_existing_managed( uow_factory: InMemoryUnitOfWorkFactory, registry_service: AgentRegistryService,