From 499629b38ba7e47f19ce3b2a1ed849dc73e02e36 Mon Sep 17 00:00:00 2001 From: 0YHR0 <97868579@qq.com> Date: Fri, 21 Aug 2026 17:44:51 +0800 Subject: [PATCH 1/7] feat(runtime): add evidence-driven reconciliation core --- src/agentmesh/api/runtime_routes.py | 73 ++- src/agentmesh/application/ports.py | 3 + .../application/runtime_reconciliation.py | 433 ++++++++++++++++++ src/agentmesh/bootstrap.py | 18 + src/agentmesh/domain/resolutions.py | 2 - src/agentmesh/domain/runtime_execution.py | 43 +- src/agentmesh/domain/tasks.py | 100 ++++ .../postgres/runtime_repositories.py | 40 +- tests/test_domain_tasks.py | 74 +++ tests/test_runtime_execution_domain.py | 41 ++ 10 files changed, 819 insertions(+), 8 deletions(-) create mode 100644 src/agentmesh/application/runtime_reconciliation.py diff --git a/src/agentmesh/api/runtime_routes.py b/src/agentmesh/api/runtime_routes.py index 6294325..3f2a4bd 100644 --- a/src/agentmesh/api/runtime_routes.py +++ b/src/agentmesh/api/runtime_routes.py @@ -1,14 +1,19 @@ from __future__ import annotations from datetime import datetime -from typing import Annotated +from typing import Annotated, Any from uuid import UUID -from fastapi import APIRouter, Depends, Query, Request -from pydantic import BaseModel +from fastapi import APIRouter, Depends, Header, Query, Request +from pydantic import BaseModel, ConfigDict from agentmesh.api.feature_routes import require_feature +from agentmesh.api.schemas import TaskResolutionResponse from agentmesh.api.security import PrincipalDependency, require_permission +from agentmesh.application.runtime_reconciliation import ( + RuntimeOutcomeReconciliationResult, + RuntimeOutcomeReconciliationService, +) from agentmesh.application.runtime_services import RuntimeRegistryService from agentmesh.domain.errors import AuthorizationDenied from agentmesh.domain.identity import Permission @@ -18,6 +23,7 @@ RuntimeVersion, ) from agentmesh.features import Feature +from agentmesh.runtime_sdk import RuntimeObservation router = APIRouter(prefix="/api/v1", tags=["runtime-control-plane"]) _dependencies = [ @@ -26,6 +32,7 @@ ] Limit = Annotated[int, Query(ge=1, le=100)] Offset = Annotated[int, Query(ge=0)] +IdempotencyKey = Annotated[str, Header(alias="Idempotency-Key", min_length=1, max_length=200)] class RuntimeRegistrationResponse(BaseModel): @@ -87,6 +94,20 @@ class RuntimeObservationResponse(BaseModel): provider_event_present: bool +class ReconcileRuntimeOutcomeRequest(BaseModel): + model_config = ConfigDict(extra="forbid") + + observation: dict[str, Any] + evidence_digest: str + evidence_reference: str + reason: str + + +class ReconcileRuntimeOutcomeResponse(BaseModel): + execution: RuntimeExecutionResponse + resolution: TaskResolutionResponse + + def _service(request: Request) -> RuntimeRegistryService: service = request.app.state.container.runtime_service if service is None: @@ -97,6 +118,18 @@ def _service(request: Request) -> RuntimeRegistryService: RuntimeServiceDependency = Annotated[RuntimeRegistryService, Depends(_service)] +def _reconciliation_service(request: Request) -> RuntimeOutcomeReconciliationService: + service = request.app.state.container.runtime_reconciliation_service + if service is None: + raise RuntimeError("Runtime reconciliation service is not configured") + return service + + +RuntimeReconciliationServiceDependency = Annotated[ + RuntimeOutcomeReconciliationService, Depends(_reconciliation_service) +] + + def _principal_uuid(principal: PrincipalDependency) -> UUID | None: try: return UUID(principal.principal_id) @@ -207,3 +240,37 @@ def list_observations( RuntimeObservationResponse(**value) for value in service.list_observations(execution_id, limit=limit, offset=offset) ] + + +@router.post( + "/runtime-executions/{execution_id}/reconcile-outcome", + response_model=ReconcileRuntimeOutcomeResponse, + dependencies=[ + *_dependencies, + Depends(require_feature(Feature.OUTCOME_RECONCILIATION)), + Depends(require_permission(Permission.OUTCOME_RECONCILE)), + ], +) +def reconcile_runtime_outcome( + execution_id: UUID, + payload: ReconcileRuntimeOutcomeRequest, + principal: PrincipalDependency, + service: RuntimeReconciliationServiceDependency, + idempotency_key: IdempotencyKey, +) -> ReconcileRuntimeOutcomeResponse: + if principal.tenant_id != service.tenant_id or not principal.authenticated: + raise AuthorizationDenied("Runtime tenant scope denied") + observation = RuntimeObservation.from_dict(payload.observation) + result: RuntimeOutcomeReconciliationResult = service.reconcile_outcome( + execution_id, + principal=principal, + observation=observation, + evidence_digest=payload.evidence_digest, + evidence_reference=payload.evidence_reference, + reason=payload.reason, + idempotency_key=idempotency_key, + ) + return ReconcileRuntimeOutcomeResponse( + execution=_execution(result.execution), + resolution=TaskResolutionResponse.from_domain(result.resolution), + ) diff --git a/src/agentmesh/application/ports.py b/src/agentmesh/application/ports.py index 19f4e99..a17b0cf 100644 --- a/src/agentmesh/application/ports.py +++ b/src/agentmesh/application/ports.py @@ -212,6 +212,9 @@ def update_observation_outcome( *, outcome: RuntimeObservationOutcome, ) -> None: ... + def find_cancel_intent( + self, execution_id: UUID, *, tenant_id: str + ) -> RuntimeLifecycleIntent | None: ... def add_lifecycle_operation(self, value: RuntimeLifecycleIntent) -> None: ... def find_lifecycle_operation( self, execution_id: UUID, *, tenant_id: str, operation_id: str diff --git a/src/agentmesh/application/runtime_reconciliation.py b/src/agentmesh/application/runtime_reconciliation.py new file mode 100644 index 0000000..a25edcb --- /dev/null +++ b/src/agentmesh/application/runtime_reconciliation.py @@ -0,0 +1,433 @@ +from __future__ import annotations + +import logging +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Any +from uuid import UUID, uuid4 + +from agentmesh.application.memory_runtime_services import RuntimeMemoryService +from agentmesh.application.ports import UnitOfWorkFactory +from agentmesh.application.research_materialization_services import ( + ResearchMaterializationService, +) +from agentmesh.domain.errors import ( + AuthorizationDenied, + IdempotencyConflict, + InvalidTaskInput, + InvalidTaskTransition, + RuntimeExecutionNotFound, +) +from agentmesh.domain.identity import PrincipalContext +from agentmesh.domain.messaging import IdempotencyRecord, MessageEnvelope +from agentmesh.domain.resolutions import TaskResolution, TaskResolutionAction +from agentmesh.domain.runtime_execution import ( + RuntimeExecution, + RuntimeExecutionPhase, + RuntimeObservationEvidence, + RuntimeObservationOutcome, +) +from agentmesh.domain.tasks import AttemptStatus, RunStatus, TaskStatus +from agentmesh.features import Feature, FeatureGateSet +from agentmesh.runtime_sdk import RuntimeObservation, RuntimePhase, canonical_digest + +logger = logging.getLogger(__name__) + +_KNOWN_TERMINAL_PHASES = { + RuntimePhase.SUCCEEDED, + RuntimePhase.FAILED, + RuntimePhase.CANCELED, + RuntimePhase.TIMED_OUT, +} + + +@dataclass(frozen=True) +class RuntimeOutcomeReconciliationResult: + execution: RuntimeExecution + resolution: TaskResolution + + +class RuntimeOutcomeReconciliationService: + """Privileged evidence-only convergence for parked managed DIRECT executions.""" + + def __init__( + self, + *, + uow_factory: UnitOfWorkFactory, + tenant_id: str, + feature_gates: FeatureGateSet, + runtime_memory_service: RuntimeMemoryService | None = None, + research_materialization_service: ResearchMaterializationService | None = None, + ) -> None: + self._uow_factory = uow_factory + self._tenant_id = tenant_id + self._feature_gates = feature_gates + self._runtime_memory_service = runtime_memory_service + self._research_materialization_service = research_materialization_service + + @property + def tenant_id(self) -> str: + return self._tenant_id + + def reconcile_outcome( + self, + execution_id: UUID, + *, + principal: PrincipalContext, + observation: RuntimeObservation, + evidence_digest: str, + evidence_reference: str, + reason: str, + idempotency_key: str, + ) -> RuntimeOutcomeReconciliationResult: + self._feature_gates.require(Feature.MANAGED_AGENT_RUNTIME) + self._feature_gates.require(Feature.OUTCOME_RECONCILIATION) + self._require_principal(principal) + normalized_reference = evidence_reference.strip() + normalized_reason = reason.strip() + normalized_key = idempotency_key.strip() + if not normalized_reference or len(normalized_reference.encode("utf-8")) > 2048: + raise InvalidTaskInput("Evidence reference must contain 1-2048 UTF-8 bytes") + if not normalized_reason or len(normalized_reason.encode("utf-8")) > 2000: + raise InvalidTaskInput("Reconciliation reason must contain 1-2000 UTF-8 bytes") + if not normalized_key: + raise IdempotencyConflict("Idempotency-Key must not be empty") + if observation.phase not in _KNOWN_TERMINAL_PHASES: + raise InvalidTaskInput("Reconciliation requires a known terminal observation") + observation_digest = canonical_digest(observation.to_dict()) + if evidence_digest != observation_digest: + raise InvalidTaskInput("Evidence digest must equal the canonical observation digest") + if UUID(observation.runtime_execution_id) != execution_id: + raise InvalidTaskInput("Observation Runtime execution identity does not match") + if observation.phase is RuntimePhase.SUCCEEDED: + if type(observation.output) is not dict or observation.usage: + raise InvalidTaskInput( + "Managed Runtime success requires mapping output and empty usage" + ) + elif observation.output is not None or observation.output_artifact_refs: + raise InvalidTaskInput("Non-success Runtime evidence cannot carry output") + + request_hash = canonical_digest( + { + "execution_id": str(execution_id), + "observation": observation.to_dict(), + "evidence_digest": evidence_digest, + "evidence_reference": normalized_reference, + "reason": normalized_reason, + } + ) + scope = ( + f"runtime-outcome-reconciliation:{self._tenant_id}:" + f"{principal.principal_id}:{execution_id}" + ) + completed_task_id: UUID | None = None + with self._uow_factory() as uow: + replay = self._existing_replay(uow, scope, normalized_key, request_hash) + if replay is not None: + return self._replay_result(uow, execution_id, replay) + + located = uow.runtimes.get_execution(execution_id, tenant_id=self._tenant_id) + if located is None: + raise RuntimeExecutionNotFound("Runtime execution was not found") + located_run = uow.runs.get(located.run_id) + if located_run is None: + raise InvalidTaskTransition("Runtime execution Run linkage was lost") + + task = uow.tasks.get(located_run.task_id, for_update=True) + run = uow.runs.get(located.run_id, for_update=True) + attempt = uow.attempts.latest_for_run(located.run_id, for_update=True) + execution = uow.runtimes.get_execution( + execution_id, tenant_id=self._tenant_id, for_update=True + ) + if task is None or task.tenant_id != self._tenant_id: + raise AuthorizationDenied("Runtime tenant scope denied") + if run is None or attempt is None or execution is None: + raise InvalidTaskTransition("Runtime reconciliation linkage was lost") + + uow.idempotency.lock(scope, normalized_key) + replay = self._existing_replay(uow, scope, normalized_key, request_hash) + if replay is not None: + return self._replay_result(uow, execution_id, replay) + self._require_parked(task, run, attempt, execution) + self._require_observation_identity(execution, observation) + self._reconcile_evidence( + uow, + execution=execution, + observation=observation, + observation_digest=observation_digest, + evidence_reference=normalized_reference, + ) + + previous_phase = execution.phase + confirmed_phase = RuntimeExecutionPhase(observation.phase.value) + reconciled_execution = execution.reconcile_terminal( + phase=confirmed_phase, + provider_sequence=observation.provider_sequence, + ) + previous_status = task.status + previous_error = task.error + action, business_reason = self._converge_business_state( + uow, + task=task, + run=run, + attempt=attempt, + execution=execution, + observation=observation, + ) + resolution = TaskResolution.create( + task_id=task.id, + action=action, + actor=principal.principal_id, + reason=normalized_reason, + previous_status=previous_status, + resulting_status=task.status, + previous_error=previous_error, + details={ + "target_type": "RUNTIME_EXECUTION", + "execution_id": str(execution.id), + "run_id": str(run.id), + "attempt_id": str(attempt.id), + "previous_phase": previous_phase.value, + "confirmed_phase": confirmed_phase.value, + "business_mapping_reason": business_reason, + "assignment_digest": execution.assignment_digest, + "observation_id": observation.observation_id, + "observation_digest": observation_digest, + "provider_event_id": observation.provider_event_id, + "snapshot_digest": observation.snapshot_digest, + "evidence_reference": normalized_reference, + }, + ) + uow.runtimes.save_execution(reconciled_execution, tenant_id=self._tenant_id) + uow.tasks.save(task) + uow.runs.save(run) + uow.attempts.save(attempt) + uow.task_resolutions.add(resolution) + uow.outbox.add( + MessageEnvelope.domain_event( + schema_name="agentmesh.runtime.outcome-reconciled", + tenant_id=self._tenant_id, + aggregate_id=task.id, + causation_id=resolution.id, + producer="agentmesh-runtime-reconciler-v1", + payload={ + "tenant_id": self._tenant_id, + "task_id": str(task.id), + "run_id": str(run.id), + "attempt_id": str(attempt.id), + "runtime_execution_id": str(execution.id), + "resolution_id": str(resolution.id), + "confirmed_phase": confirmed_phase.value, + }, + ) + ) + uow.idempotency.add( + IdempotencyRecord.create( + scope=scope, + key=normalized_key, + request_hash=request_hash, + result={"resolution_id": str(resolution.id)}, + ) + ) + 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() + completed_task_id = task.id if task.status is TaskStatus.COMPLETED else None + result = RuntimeOutcomeReconciliationResult(reconciled_execution, resolution) + + if completed_task_id is not None and self._research_materialization_service is not None: + try: + self._research_materialization_service.materialize_if_ready( + completed_task_id, actor=principal.principal_id + ) + except Exception: + logger.warning( + "Automatic research materialization failed for reconciled Task %s", + completed_task_id, + exc_info=True, + ) + return result + + def _require_principal(self, principal: PrincipalContext) -> None: + if not principal.authenticated or principal.tenant_id != self._tenant_id: + raise AuthorizationDenied( + "Runtime outcome reconciliation requires an authenticated tenant Principal" + ) + + @staticmethod + def _existing_replay(uow: Any, scope: str, key: str, request_hash: str) -> dict | None: + record = uow.idempotency.get(scope, key) + if record is None: + return None + if record.request_hash != request_hash: + raise IdempotencyConflict("Idempotency key was reused with a different request") + return record.result + + def _replay_result( + self, uow: Any, execution_id: UUID, replay: dict + ) -> RuntimeOutcomeReconciliationResult: + execution = uow.runtimes.get_execution(execution_id, tenant_id=self._tenant_id) + resolution = uow.task_resolutions.get(UUID(str(replay["resolution_id"]))) + if execution is None or resolution is None: + raise InvalidTaskTransition("Reconciliation idempotency result was lost") + return RuntimeOutcomeReconciliationResult(execution, resolution) + + @staticmethod + def _require_parked(task: Any, run: Any, attempt: Any, execution: RuntimeExecution) -> None: + if ( + task.status is not TaskStatus.RECONCILIATION_REQUIRED + or run.status is not RunStatus.RECONCILIATION_REQUIRED + or attempt.status is not AttemptStatus.OUTCOME_UNKNOWN + or run.runtime_authority != "managed" + or task.current_run_id != run.id + or execution.run_id != run.id + or execution.current_owner_attempt_id != attempt.id + or execution.current_fencing_token != attempt.fencing_token + or execution.phase + not in {RuntimeExecutionPhase.OUTCOME_UNKNOWN, RuntimeExecutionPhase.LOST} + ): + raise InvalidTaskTransition( + "Runtime execution is not a strictly consistent parked managed Run" + ) + + @staticmethod + def _require_observation_identity( + execution: RuntimeExecution, observation: RuntimeObservation + ) -> None: + if ( + UUID(observation.assignment_id) != execution.assignment_id + or observation.assignment_digest != execution.assignment_digest + ): + raise InvalidTaskInput("Observation assignment identity does not match") + + @staticmethod + def _reconcile_evidence( + uow: Any, + *, + execution: RuntimeExecution, + observation: RuntimeObservation, + observation_digest: str, + evidence_reference: str, + ) -> None: + prior = uow.runtimes.prior_observations( + execution.id, + tenant_id=execution.tenant_id, + observation_id=observation.observation_id, + digest=observation_digest, + ) + if any( + item.observation_id == observation.observation_id + and item.observation_digest != observation_digest + for item in prior + ): + raise InvalidTaskTransition("Observation identity conflicts with existing evidence") + exact = next( + ( + item + for item in prior + if item.observation_id == observation.observation_id + and item.observation_digest == observation_digest + ), + None, + ) + expected_provider = { + "provider_event_id": observation.provider_event_id, + "snapshot_digest": observation.snapshot_digest, + } + if exact is not None: + actual_provider = { + "provider_event_id": exact.evidence.get("provider_event_id"), + "snapshot_digest": exact.evidence.get("snapshot_digest"), + } + if ( + exact.runtime_execution_id != execution.id + or exact.assignment_id != execution.assignment_id + or exact.assignment_digest != execution.assignment_digest + or exact.phase is not RuntimeExecutionPhase(observation.phase.value) + or actual_provider != expected_provider + or exact.processing_outcome + not in {RuntimeObservationOutcome.CONFLICT, RuntimeObservationOutcome.RECONCILED} + ): + raise InvalidTaskTransition("Existing Runtime evidence cannot be reconciled") + if exact.processing_outcome is RuntimeObservationOutcome.CONFLICT: + uow.runtimes.update_observation_outcome( + exact, outcome=RuntimeObservationOutcome.RECONCILED + ) + return + uow.runtimes.add_observation( + RuntimeObservationEvidence( + id=uuid4(), + tenant_id=execution.tenant_id, + runtime_execution_id=execution.id, + observation_id=observation.observation_id, + observation_digest=observation_digest, + assignment_id=execution.assignment_id, + assignment_digest=execution.assignment_digest, + provider_sequence=observation.provider_sequence, + phase=RuntimeExecutionPhase(observation.phase.value), + observed_at=observation.observed_at.astimezone(timezone.utc), + received_at=datetime.now(timezone.utc), + safe_summary="Operator-confirmed Runtime outcome", + processing_outcome=RuntimeObservationOutcome.RECONCILED, + provider_event_present=observation.provider_event_id is not None, + evidence={ + **expected_provider, + "evidence_reference": evidence_reference, + }, + ) + ) + + @staticmethod + def _converge_business_state( + uow: Any, + *, + task: Any, + run: Any, + attempt: Any, + execution: RuntimeExecution, + observation: RuntimeObservation, + ) -> tuple[TaskResolutionAction, str]: + if observation.phase is RuntimePhase.SUCCEEDED: + output = dict(observation.output) + deadline_exceeded = ( + task.budget is not None + and task.budget.deadline is not None + and observation.observed_at.astimezone(timezone.utc) + >= task.budget.deadline.astimezone(timezone.utc) + ) + run.reconcile_runtime_succeeded(output) + attempt.reconcile_runtime_succeeded() + task.reconcile_runtime_succeeded( + run.id, output, budget_deadline_exceeded=deadline_exceeded + ) + return ( + TaskResolutionAction.RECONCILE_RUNTIME_SUCCEEDED, + "budget_deadline_exceeded" if deadline_exceeded else "runtime.confirmed_success", + ) + if observation.phase is RuntimePhase.CANCELED: + cancel_intent = uow.runtimes.find_cancel_intent( + execution.id, tenant_id=execution.tenant_id + ) + if cancel_intent is not None: + run.reconcile_runtime_canceled("runtime.reconciled_canceled") + attempt.reconcile_runtime_canceled("runtime.reconciled_canceled") + task.reconcile_runtime_canceled(run.id, "runtime.reconciled_canceled") + return ( + TaskResolutionAction.RECONCILE_RUNTIME_CANCELED, + "runtime.reconciled_canceled", + ) + reason = "runtime.unrequested_cancellation" + run.reconcile_runtime_failed(reason) + attempt.reconcile_runtime_failed(reason) + task.reconcile_runtime_failed(run.id, reason) + return TaskResolutionAction.RECONCILE_RUNTIME_CANCELED, reason + if observation.phase is RuntimePhase.TIMED_OUT: + reason = "runtime.reconciled_timed_out" + action = TaskResolutionAction.RECONCILE_RUNTIME_TIMED_OUT + else: + reason = "runtime.reconciled_failed" + action = TaskResolutionAction.RECONCILE_RUNTIME_FAILED + run.reconcile_runtime_failed(reason) + attempt.reconcile_runtime_failed(reason) + task.reconcile_runtime_failed(run.id, reason) + return action, reason diff --git a/src/agentmesh/bootstrap.py b/src/agentmesh/bootstrap.py index eb64f0d..ef558c5 100644 --- a/src/agentmesh/bootstrap.py +++ b/src/agentmesh/bootstrap.py @@ -46,6 +46,9 @@ ResearchMaterializationService, ) from agentmesh.application.resolution_services import TaskResolutionService +from agentmesh.application.runtime_reconciliation import ( + RuntimeOutcomeReconciliationService, +) from agentmesh.application.runtime_services import RuntimeRegistryService from agentmesh.application.services import RunExecutionService, TaskApplicationService from agentmesh.application.tool_services import ToolInvocationService @@ -147,6 +150,7 @@ class ApplicationContainer: extension_runtime: ExtensionRuntime mcp_catalog_client: OfficialMcpRegistryClient | None = None runtime_service: RuntimeRegistryService | None = None + runtime_reconciliation_service: RuntimeOutcomeReconciliationService | None = None event_stream: RedisDomainEventStream | None = None close_callback: Callable[[], None] = lambda: None @@ -425,6 +429,19 @@ def build_api_container(settings: Settings | None = None) -> ApplicationContaine artifact_service=artifact_service, tenant_id=runtime_settings.tenant_id, ) + runtime_memory_service = RuntimeMemoryService( + uow_factory=uow_factory, + memory_service=organizational_memory_service, + tenant_id=runtime_settings.tenant_id, + feature_gates=feature_gates, + ) + runtime_reconciliation_service = RuntimeOutcomeReconciliationService( + uow_factory=uow_factory, + tenant_id=runtime_settings.tenant_id, + feature_gates=feature_gates, + runtime_memory_service=runtime_memory_service, + research_materialization_service=research_materialization_service, + ) extension_runtime = ExtensionRuntime.load( RUNTIME_EXTENSION_REGISTRY, ExtensionContext( @@ -484,6 +501,7 @@ def close() -> None: extension_runtime=extension_runtime, mcp_catalog_client=OfficialMcpRegistryClient(), runtime_service=runtime_service, + runtime_reconciliation_service=runtime_reconciliation_service, event_stream=event_stream, close_callback=close, ) diff --git a/src/agentmesh/domain/resolutions.py b/src/agentmesh/domain/resolutions.py index bc8c8a4..3c09c69 100644 --- a/src/agentmesh/domain/resolutions.py +++ b/src/agentmesh/domain/resolutions.py @@ -18,8 +18,6 @@ class TaskResolutionAction(str, Enum): RECONCILE_MCP_FAILED = "RECONCILE_MCP_FAILED" BIND_A2A_REMOTE_TASK = "BIND_A2A_REMOTE_TASK" RECONCILE_A2A_NOT_DELIVERED = "RECONCILE_A2A_NOT_DELIVERED" - # Expand-phase reader compatibility for managed Runtime reconciliation. - # No command writes these actions until the follow-up writer rollout. RECONCILE_RUNTIME_SUCCEEDED = "RECONCILE_RUNTIME_SUCCEEDED" RECONCILE_RUNTIME_FAILED = "RECONCILE_RUNTIME_FAILED" RECONCILE_RUNTIME_CANCELED = "RECONCILE_RUNTIME_CANCELED" diff --git a/src/agentmesh/domain/runtime_execution.py b/src/agentmesh/domain/runtime_execution.py index 31a2e42..b8567dd 100644 --- a/src/agentmesh/domain/runtime_execution.py +++ b/src/agentmesh/domain/runtime_execution.py @@ -85,8 +85,6 @@ class RuntimeObservationOutcome(str, Enum): GAP = "GAP" STALE_OWNER = "STALE_OWNER" CONFLICT = "CONFLICT" - # Expand-phase reader compatibility for A4.1b.2. Writer activation is a - # separate rollout after every old reader has been replaced. RECONCILED = "RECONCILED" @@ -655,6 +653,47 @@ def apply_observation( terminal_at=timestamp if phase.terminal else self.terminal_at, ) + def reconcile_terminal( + self, + *, + phase: RuntimeExecutionPhase, + provider_sequence: int | None, + now: datetime | None = None, + ) -> RuntimeExecution: + """Converge an ambiguous execution from independently verified evidence.""" + if self.phase not in { + RuntimeExecutionPhase.OUTCOME_UNKNOWN, + RuntimeExecutionPhase.LOST, + }: + raise InvalidTaskTransition( + "Only an ambiguous Runtime execution can be reconciled" + ) + if phase not in { + RuntimeExecutionPhase.SUCCEEDED, + RuntimeExecutionPhase.FAILED, + RuntimeExecutionPhase.CANCELED, + RuntimeExecutionPhase.TIMED_OUT, + }: + raise InvalidTaskTransition( + "Runtime reconciliation requires a known terminal phase" + ) + if provider_sequence is not None and self.provider_sequence is not None: + if provider_sequence < self.provider_sequence: + raise InvalidTaskTransition( + "Runtime reconciliation cannot regress provider sequence" + ) + timestamp = now or utc_now() + return replace( + self, + phase=phase, + provider_sequence=( + provider_sequence if provider_sequence is not None else self.provider_sequence + ), + version=self.version + 1, + updated_at=timestamp, + terminal_at=timestamp, + ) + def _freeze_json(value: Any) -> Any: if type(value) is dict: diff --git a/src/agentmesh/domain/tasks.py b/src/agentmesh/domain/tasks.py index 11eb3e6..5ff4fcb 100644 --- a/src/agentmesh/domain/tasks.py +++ b/src/agentmesh/domain/tasks.py @@ -391,6 +391,57 @@ def require_runtime_reconciliation(self, run_id: UUID, reason: str) -> None: self.error = _runtime_reconciliation_reason(reason) self._touch() + def reconcile_runtime_succeeded( + self, + run_id: UUID, + output: dict[str, Any], + *, + budget_deadline_exceeded: bool = False, + ) -> None: + self._require_active_run( + run_id, + "reconcile Runtime success", + expected=TaskStatus.RECONCILIATION_REQUIRED, + ) + if self.execution_mode is not TaskExecutionMode.DIRECT: + raise InvalidTaskTransition("Only direct Tasks can reconcile Runtime outcomes") + if budget_deadline_exceeded: + self.status = TaskStatus.WAITING_APPROVAL + self.current_run_id = None + self.output = None + self.candidate_output = dict(output) + self.error = "budget_deadline_exceeded" + self.budget_exhausted_reason = "budget_deadline_exceeded" + else: + self.status = TaskStatus.COMPLETED + self.output = dict(output) + self.candidate_output = None + self.error = None + self.budget_exhausted_reason = None + self._touch() + + def reconcile_runtime_failed(self, run_id: UUID, reason: str) -> None: + self._reconcile_runtime_terminal(run_id, TaskStatus.FAILED, reason) + + def reconcile_runtime_canceled(self, run_id: UUID, reason: str) -> None: + self._reconcile_runtime_terminal(run_id, TaskStatus.CANCELED, reason) + + def _reconcile_runtime_terminal( + self, run_id: UUID, status: TaskStatus, reason: str + ) -> None: + self._require_active_run( + run_id, + "reconcile Runtime outcome", + expected=TaskStatus.RECONCILIATION_REQUIRED, + ) + if self.execution_mode is not TaskExecutionMode.DIRECT: + raise InvalidTaskTransition("Only direct Tasks can reconcile Runtime outcomes") + normalized = _runtime_reconciliation_reason(reason) + self.status = status + self.output = None + self.error = normalized + self._touch() + def start_coordination(self) -> None: self._require_status(TaskStatus.CREATED, "start coordination") if self.execution_mode != TaskExecutionMode.COORDINATED: @@ -869,6 +920,31 @@ def require_runtime_reconciliation(self, reason: str) -> None: self.output = None self.error = _runtime_reconciliation_reason(reason) + def reconcile_runtime_succeeded(self, output: dict[str, Any]) -> None: + self._require_reconciliation("reconcile Runtime success") + self.status = RunStatus.SUCCEEDED + self.output = dict(output) + self.error = None + self.completed_at = utc_now() + + def reconcile_runtime_failed(self, reason: str) -> None: + self._reconcile_runtime_terminal(RunStatus.FAILED, reason) + + def reconcile_runtime_canceled(self, reason: str) -> None: + self._reconcile_runtime_terminal(RunStatus.CANCELED, reason) + + def _reconcile_runtime_terminal(self, status: RunStatus, reason: str) -> None: + self._require_reconciliation("reconcile Runtime outcome") + self.status = status + self.output = None + self.error = _runtime_reconciliation_reason(reason) + self.completed_at = utc_now() + + def _require_reconciliation(self, action: str) -> None: + if self.runtime_authority != "managed": + raise InvalidTaskTransition("Only managed Runs can reconcile Runtime outcomes") + self._require_status(RunStatus.RECONCILIATION_REQUIRED, action) + def wait_for_remote(self) -> None: self._require_status(RunStatus.QUEUED, "wait for remote") self.status = RunStatus.WAITING_REMOTE @@ -1075,6 +1151,30 @@ def mark_outcome_unknown(self, reason: str) -> None: self.error = _runtime_reconciliation_reason(reason) self.completed_at = utc_now() + def reconcile_runtime_succeeded(self) -> None: + self._require_outcome_unknown("reconcile Runtime success") + self.status = AttemptStatus.SUCCEEDED + self.error = None + self.completed_at = utc_now() + + def reconcile_runtime_failed(self, reason: str) -> None: + self._reconcile_runtime_terminal(AttemptStatus.FAILED, reason) + + def reconcile_runtime_canceled(self, reason: str) -> None: + self._reconcile_runtime_terminal(AttemptStatus.CANCELED, reason) + + def _reconcile_runtime_terminal(self, status: AttemptStatus, reason: str) -> None: + self._require_outcome_unknown("reconcile Runtime outcome") + self.status = status + self.error = _runtime_reconciliation_reason(reason) + self.completed_at = utc_now() + + def _require_outcome_unknown(self, action: str) -> None: + if self.status is not AttemptStatus.OUTCOME_UNKNOWN: + raise InvalidTaskTransition( + f"Cannot {action} attempt {self.id} from status {self.status.value}" + ) + def renew( self, *, diff --git a/src/agentmesh/infrastructure/postgres/runtime_repositories.py b/src/agentmesh/infrastructure/postgres/runtime_repositories.py index ffbefa7..88e861f 100644 --- a/src/agentmesh/infrastructure/postgres/runtime_repositories.py +++ b/src/agentmesh/infrastructure/postgres/runtime_repositories.py @@ -478,7 +478,11 @@ def add_observation(self, value: RuntimeObservationEvidence) -> None: observation_digest=value.observation_digest, assignment_id=value.assignment_id, assignment_digest=value.assignment_digest, - provider_event_id=None, + provider_event_id=( + value.evidence.get("provider_event_id") + if type(value.evidence.get("provider_event_id")) is str + else None + ), provider_sequence=value.provider_sequence, phase=value.phase.value, observed_at=value.observed_at, @@ -560,6 +564,39 @@ def update_observation_outcome( raise LookupError(value.id) record.processing_outcome = outcome.value + def find_cancel_intent( + self, execution_id: UUID, *, tenant_id: str + ) -> RuntimeLifecycleIntent | None: + record = self._session.scalar( + select(RuntimeLifecycleOperationRecord) + .join( + RuntimeExecutionRecord, + RuntimeExecutionRecord.id + == RuntimeLifecycleOperationRecord.runtime_execution_id, + ) + .join(TaskRunRecord, TaskRunRecord.id == RuntimeExecutionRecord.run_id) + .join(TaskRecord, TaskRecord.id == TaskRunRecord.task_id) + .where( + RuntimeLifecycleOperationRecord.runtime_execution_id == execution_id, + RuntimeLifecycleOperationRecord.tenant_id == tenant_id, + TaskRecord.tenant_id == tenant_id, + RuntimeLifecycleOperationRecord.operation + == RuntimeLifecycleOperation.CANCEL.value, + RuntimeLifecycleOperationRecord.status.in_( + [ + RuntimeLifecycleStatus.REQUESTED.value, + RuntimeLifecycleStatus.ACCEPTED.value, + ] + ), + ) + .order_by( + RuntimeLifecycleOperationRecord.created_at.desc(), + RuntimeLifecycleOperationRecord.id.desc(), + ) + .limit(1) + ) + return _lifecycle_projection(record) + def add_lifecycle_operation(self, value: RuntimeLifecycleIntent) -> None: self._session.add( RuntimeLifecycleOperationRecord( @@ -762,6 +799,7 @@ def _observation_projection(record: RuntimeObservationRecord) -> RuntimeObservat safe_summary=record.safe_summary, processing_outcome=RuntimeObservationOutcome(record.processing_outcome), provider_event_present=record.provider_event_id is not None, + evidence=dict(record.evidence), ) diff --git a/tests/test_domain_tasks.py b/tests/test_domain_tasks.py index 4cb5c7b..e12cb88 100644 --- a/tests/test_domain_tasks.py +++ b/tests/test_domain_tasks.py @@ -126,6 +126,80 @@ def test_runtime_reconciliation_state_is_fail_closed() -> None: ).mark_outcome_unknown("x" * 513) +def _parked_managed_direct(): + task = Task.create( + tenant_id="test", + objective="Converge evidence", + 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.unknown") + run.require_runtime_reconciliation("runtime.unknown") + attempt.mark_outcome_unknown("runtime.unknown") + return task, run, attempt + + +def test_parked_managed_direct_has_dedicated_success_exit() -> None: + task, run, attempt = _parked_managed_direct() + task.candidate_output = {"stale": True} + task.budget_exhausted_reason = "stale" + + run.reconcile_runtime_succeeded({"answer": 42}) + attempt.reconcile_runtime_succeeded() + task.reconcile_runtime_succeeded(run.id, {"answer": 42}) + + assert task.status is TaskStatus.COMPLETED + assert task.output == {"answer": 42} + assert task.candidate_output is None + assert task.budget_exhausted_reason is None + assert run.status is RunStatus.SUCCEEDED + assert attempt.status is AttemptStatus.SUCCEEDED + assert attempt.completed_at is not None + + +def test_parked_success_at_budget_deadline_waits_for_approval() -> None: + task, run, attempt = _parked_managed_direct() + + run.reconcile_runtime_succeeded({"answer": 42}) + attempt.reconcile_runtime_succeeded() + task.reconcile_runtime_succeeded( + run.id, {"answer": 42}, budget_deadline_exceeded=True + ) + + assert task.status is TaskStatus.WAITING_APPROVAL + assert task.current_run_id is None + assert task.output is None + assert task.candidate_output == {"answer": 42} + assert task.error == task.budget_exhausted_reason == "budget_deadline_exceeded" + + +def test_dedicated_runtime_reconciliation_exits_reject_ordinary_states() -> None: + task, run, attempt = _parked_managed_direct() + task.reconcile_runtime_failed(run.id, "runtime.failed") + run.reconcile_runtime_failed("runtime.failed") + attempt.reconcile_runtime_failed("runtime.failed") + assert task.status is TaskStatus.FAILED + assert run.status is RunStatus.FAILED + assert attempt.status is AttemptStatus.FAILED + assert attempt.completed_at is not None + with pytest.raises(InvalidTaskTransition): + attempt.reconcile_runtime_succeeded() + + 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_runtime_execution_domain.py b/tests/test_runtime_execution_domain.py index e2a64ae..f003341 100644 --- a/tests/test_runtime_execution_domain.py +++ b/tests/test_runtime_execution_domain.py @@ -41,6 +41,47 @@ def test_observation_preserves_sequence_when_provider_omits_it() -> None: assert value.provider_sequence == 1 +@pytest.mark.parametrize( + "phase", + [ + RuntimeExecutionPhase.SUCCEEDED, + RuntimeExecutionPhase.FAILED, + RuntimeExecutionPhase.CANCELED, + RuntimeExecutionPhase.TIMED_OUT, + ], +) +def test_ambiguous_execution_has_a_dedicated_terminal_reconciliation_exit( + phase: RuntimeExecutionPhase, +) -> None: + ambiguous = _execution().apply_observation( + phase=RuntimeExecutionPhase.OUTCOME_UNKNOWN, provider_sequence=1 + ) + + reconciled = ambiguous.reconcile_terminal(phase=phase, provider_sequence=2) + + assert reconciled.phase is phase + assert reconciled.provider_sequence == 2 + assert reconciled.terminal_at is not None + with pytest.raises(InvalidTaskTransition): + reconciled.reconcile_terminal( + phase=RuntimeExecutionPhase.FAILED, provider_sequence=3 + ) + + +def test_runtime_reconciliation_rejects_non_ambiguous_or_nonterminal_transitions() -> None: + with pytest.raises(InvalidTaskTransition): + _execution().reconcile_terminal( + phase=RuntimeExecutionPhase.SUCCEEDED, provider_sequence=None + ) + ambiguous = _execution().apply_observation( + phase=RuntimeExecutionPhase.LOST, provider_sequence=None + ) + with pytest.raises(InvalidTaskTransition): + ambiguous.reconcile_terminal( + phase=RuntimeExecutionPhase.RUNNING, provider_sequence=None + ) + + def test_lifecycle_receipt_summary_is_an_immutable_json_projection() -> None: receipt = {"status": "accepted", "details": {"attempt": 1}} value = RuntimeLifecycleIntent( From 2c71df32f178d9c553fa1b517604f82483cfadcf Mon Sep 17 00:00:00 2001 From: 0YHR0 <97868579@qq.com> Date: Fri, 21 Aug 2026 17:47:14 +0800 Subject: [PATCH 2/7] test(runtime): cover reconciliation API and downgrade --- tests/test_runtime_routes.py | 127 +++++++++++++++++++++++++++++++++++ 1 file changed, 127 insertions(+) diff --git a/tests/test_runtime_routes.py b/tests/test_runtime_routes.py index 7610e8f..e75eae3 100644 --- a/tests/test_runtime_routes.py +++ b/tests/test_runtime_routes.py @@ -2,6 +2,7 @@ from uuid import UUID, uuid4 import pytest +from fastapi.encoders import jsonable_encoder from fastapi.testclient import TestClient from agentmesh.api.app import create_app @@ -12,6 +13,9 @@ list_versions, ) from agentmesh.api.security import get_principal_context +from agentmesh.application.runtime_reconciliation import ( + RuntimeOutcomeReconciliationResult, +) from agentmesh.application.runtime_services import RuntimeRegistryService from agentmesh.domain.errors import ( FeatureDisabled, @@ -19,6 +23,7 @@ RuntimeRegistryConflict, ) from agentmesh.domain.identity import PrincipalContext, PrincipalType, Role +from agentmesh.domain.resolutions import TaskResolution, TaskResolutionAction from agentmesh.domain.runtime_execution import ( RuntimeExecution, RuntimeExecutionPhase, @@ -29,7 +34,9 @@ RuntimeVersionStatus, RuntimeVisibility, ) +from agentmesh.domain.tasks import TaskStatus from agentmesh.features import FeatureGateSet +from agentmesh.runtime_sdk import RuntimeObservation, RuntimePhase, canonical_digest def _principal( @@ -145,6 +152,27 @@ def get_execution(self, execution_id): raise RuntimeExecutionNotFound("missing") +class _ReconciliationService: + tenant_id = "test-tenant" + + def __init__(self, execution: RuntimeExecution) -> None: + self.execution = execution + self.calls = [] + + def reconcile_outcome(self, execution_id, **kwargs): + self.calls.append((execution_id, kwargs)) + resolution = TaskResolution.create( + task_id=self.execution.run_id, + action=TaskResolutionAction.RECONCILE_RUNTIME_SUCCEEDED, + actor=kwargs["principal"].principal_id, + reason=kwargs["reason"], + previous_status=TaskStatus.RECONCILIATION_REQUIRED, + resulting_status=TaskStatus.COMPLETED, + previous_error="runtime.unknown", + ) + return RuntimeOutcomeReconciliationResult(self.execution, resolution) + + def test_runtime_routes_use_authenticated_principal_and_redact_opaque_refs() -> None: service = _ProjectionService() principal = _principal(service.tenant_id) @@ -239,3 +267,102 @@ def test_runtime_http_feature_and_rbac_dependencies_are_not_bypassed(application ) with TestClient(no_permission) as client: assert client.get("/api/v1/runtimes").status_code == 403 + + +def test_runtime_reconciliation_http_requires_gate_permission_and_idempotency( + application_container, +) -> None: + projection = _ProjectionService() + execution = projection.execution.apply_observation( + phase=RuntimeExecutionPhase.OUTCOME_UNKNOWN, + provider_sequence=2, + ) + service = _ReconciliationService(execution) + application_container.runtime_service = projection + application_container.runtime_reconciliation_service = service + application_container.feature_gates = FeatureGateSet.from_config( + "full", + "managed_agent_runtime=true,outcome_reconciliation=true,identity_rbac=true", + ) + principal = _principal(service.tenant_id) + observation = RuntimeObservation( + observation_id=str(uuid4()), + runtime_execution_id=str(execution.id), + assignment_id=str(execution.assignment_id), + assignment_digest=execution.assignment_digest, + phase=RuntimePhase.SUCCEEDED, + observed_at=datetime.now(timezone.utc), + provider_event_id="operator-evidence-1", + output={"answer": 42}, + ) + payload = { + "observation": jsonable_encoder(observation.to_dict()), + "evidence_digest": canonical_digest(observation.to_dict()), + "evidence_reference": "case://runtime/42", + "reason": "Provider support confirmed completion", + } + application = create_app(application_container) + application.dependency_overrides[get_principal_context] = lambda: principal + with TestClient(application) as client: + missing_key = client.post( + f"/api/v1/runtime-executions/{execution.id}/reconcile-outcome", json=payload + ) + accepted = client.post( + f"/api/v1/runtime-executions/{execution.id}/reconcile-outcome", + json=payload, + headers={"Idempotency-Key": "runtime-reconcile-1"}, + ) + assert missing_key.status_code == 422 + assert accepted.status_code == 200 + assert accepted.json()["resolution"]["action"] == "RECONCILE_RUNTIME_SUCCEEDED" + assert service.calls[0][1]["idempotency_key"] == "runtime-reconcile-1" + + +def test_runtime_reconciliation_http_rejects_cross_tenant_and_missing_permission( + application_container, +) -> None: + projection = _ProjectionService() + service = _ReconciliationService(projection.execution) + application_container.runtime_service = projection + application_container.runtime_reconciliation_service = service + application_container.feature_gates = FeatureGateSet.from_config( + "full", + "managed_agent_runtime=true,outcome_reconciliation=true,identity_rbac=true", + ) + application = create_app(application_container) + observation = RuntimeObservation( + observation_id=str(uuid4()), + runtime_execution_id=str(projection.execution.id), + assignment_id=str(projection.execution.assignment_id), + assignment_digest=projection.execution.assignment_digest, + phase=RuntimePhase.FAILED, + observed_at=datetime.now(timezone.utc), + snapshot_digest="a" * 64, + ) + payload = { + "observation": jsonable_encoder(observation.to_dict()), + "evidence_digest": canonical_digest(observation.to_dict()), + "evidence_reference": "case://runtime/failure", + "reason": "Confirmed failure", + } + application.dependency_overrides[get_principal_context] = lambda: _principal( + "another-tenant" + ) + with TestClient(application) as client: + cross_tenant = client.post( + f"/api/v1/runtime-executions/{projection.execution.id}/reconcile-outcome", + json=payload, + headers={"Idempotency-Key": "cross-tenant"}, + ) + application.dependency_overrides[get_principal_context] = lambda: _principal( + service.tenant_id, frozenset({Role.AGENT_AUTHOR}) + ) + with TestClient(application) as client: + denied = client.post( + f"/api/v1/runtime-executions/{projection.execution.id}/reconcile-outcome", + json=payload, + headers={"Idempotency-Key": "denied"}, + ) + assert cross_tenant.status_code == 403 + assert denied.status_code == 403 + assert service.calls == [] From 0351e93c099f4037298d6cb51a1bb64c071d1985 Mon Sep 17 00:00:00 2001 From: 0YHR0 <97868579@qq.com> Date: Sun, 23 Aug 2026 10:41:23 +0800 Subject: [PATCH 3/7] test(runtime): prove atomic outcome reconciliation --- .../modules/runtime-outcome-reconciliation.md | 44 ++ docs/implementation-status.md | 25 +- .../runtime-direct-cutover-rollback.md | 16 +- docs/roadmap.md | 8 +- .../test_managed_direct_worker_postgres.py | 668 +++++++++++++++++- tests/test_runtime_reconciliation_service.py | 112 +++ 6 files changed, 860 insertions(+), 13 deletions(-) create mode 100644 docs/architecture/modules/runtime-outcome-reconciliation.md create mode 100644 tests/test_runtime_reconciliation_service.py diff --git a/docs/architecture/modules/runtime-outcome-reconciliation.md b/docs/architecture/modules/runtime-outcome-reconciliation.md new file mode 100644 index 0000000..6ee5d51 --- /dev/null +++ b/docs/architecture/modules/runtime-outcome-reconciliation.md @@ -0,0 +1,44 @@ +# Runtime outcome reconciliation + +This module is the evidence-driven, privileged exit for a managed DIRECT execution parked after +the provider dispatch boundary. It converges known provider evidence without executing the Agent +again. + +## Admission and evidence + +The API requires the managed-runtime and outcome-reconciliation feature gates, the +`outcome:reconcile` permission, an authenticated same-tenant Principal, and an idempotency key. It +does not require the direct-cutover gate because disabling new admission must not strand existing +work. + +The request contains a complete public `RuntimeObservation`, its canonical digest, a bounded +evidence reference, and a bounded operator reason. Only `SUCCEEDED`, `FAILED`, `CANCELED`, and +`TIMED_OUT` observations are accepted. Execution, assignment, digest, phase, and provider identity +evidence must match the persisted Runtime execution. Success remains limited to mapping output and +empty usage; other terminal phases cannot carry successful output. + +## Atomic convergence + +After locating the execution without a lock, the service locks Task, Run, latest Attempt, then +RuntimeExecution and revalidates the complete parked quartet. In one UoW it records or reuses exact +immutable observation evidence, reconciles Runtime and business state, adds a TaskResolution and +`agentmesh.runtime.outcome-reconciled` Outbox event, and stores the idempotency result. It never +holds a provider call inside a transaction because it never calls a provider at all. + +Exact replay returns the existing resolution. A different request using the same key, conflicting +evidence, stale ownership, or a concurrently settled execution fails closed. Competing operators +therefore have one committed winner. + +Confirmed success at or after the UTC budget deadline leaves Runtime, Run, and Attempt succeeded +but places the Task in `WAITING_APPROVAL` with candidate output. Parking already settled budget and +released quota, so reconciliation does not repeat those operations. A confirmed cancellation only +maps Task/Run/Attempt to canceled when a persisted cancel intent exists; otherwise the Runtime is +canceled and the business objects fail with `runtime.unrequested_cancellation`. + +## Rollback boundary + +The writer uses reader/schema compatibility delivered by A4.1b.2a. Once new observation or +resolution values have been written, migration 0048 is the schema floor. Operators may disable the +writer gate and roll the application back to the compatibility release, but must not downgrade to +0047. This slice does not provide reviewed/coordinated cutover, generic subprocess authority, or +production durable reattach. diff --git a/docs/implementation-status.md b/docs/implementation-status.md index 684b7b8..4cbc01b 100644 --- a/docs/implementation-status.md +++ b/docs/implementation-status.md @@ -1,7 +1,7 @@ # Implementation status Status: Alpha baseline -Last updated: 2026-08-21 +Last updated: 2026-08-23 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 @@ -150,6 +150,29 @@ A4.1b.2a reconciliation reader compatibility: new value, 0048 becomes the schema floor; application rollback targets this compatibility release rather than 0047. +A4.1b.2b evidence-driven outcome reconciliation (writer slice): + +- A privileged `POST /api/v1/runtime-executions/{execution_id}/reconcile-outcome` command accepts + a canonical, identity-bound terminal `RuntimeObservation`; it requires the managed-runtime and + outcome-reconciliation gates, `outcome:reconcile`, an authenticated same-tenant Principal, and + an `Idempotency-Key`. It deliberately does not depend on the new-Run cutover gate. +- The command is the only exit from the managed DIRECT reconciliation states. It locks and + revalidates the Task/Run/latest Attempt/Runtime quartet, never calls or redispatches a provider, + and commits evidence, Runtime and business convergence, TaskResolution, Outbox, and idempotency + in one UoW. Exact replay is side-effect free; conflicting evidence fails closed. +- Confirmed success retains the A4.1 mapping-output/empty-usage limit. An observation at or after + the pinned deadline preserves the provider success but moves the Task to `WAITING_APPROVAL` with + candidate output. Confirmed cancellation maps business objects to `CANCELED` only when a + persisted cancel intent exists; otherwise they fail as `runtime.unrequested_cancellation`. +- Parking already conservatively settles budget and releases quota, so reconciliation never does + either twice. Completed-task memory capture remains transactional and research materialization + remains post-commit best effort. PostgreSQL coverage exercises atomic convergence and rollback, + competing conclusions, replay, pre-existing evidence, stale fencing, and zero redispatch. +- This remains CI/test-only and disabled by default. Once a writer stores 0048-only values, schema + 0048 is the rollback floor; roll back the application to the deployed b.2a compatibility release, + not to a pre-0048 reader. Reviewed/coordinated authority and production durable reattach remain + outside this slice. + ## Current runnable baseline AgentMesh currently provides durable direct, independently reviewed, and coordinated Subtask DAG diff --git a/docs/operations/runtime-direct-cutover-rollback.md b/docs/operations/runtime-direct-cutover-rollback.md index 9ace22f..de23ced 100644 --- a/docs/operations/runtime-direct-cutover-rollback.md +++ b/docs/operations/runtime-direct-cutover-rollback.md @@ -1,8 +1,9 @@ # Deterministic direct-runtime admission rollback -Last updated: 2026-08-21 +Last updated: 2026-08-23 -This runbook describes the A4.1a/A4.1b.1 CI/test-only admission and managed DIRECT Worker path. +This runbook describes the A4.1a/A4.1b CI/test-only admission, managed DIRECT Worker, and +evidence-driven reconciliation path. It is not a production runtime cutover procedure. ## Scope @@ -29,13 +30,14 @@ immutable persisted snapshot. 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. +5. Use only the gated, permission-protected runtime outcome reconciliation command to converge a + `RECONCILIATION_REQUIRED` execution from canonical provider evidence. The command remains + available when direct-cutover admission is off and never redispatches the provider. Manual + status edits, direct database repair, and blind provider retry are prohibited. Migration 0047 keeps legacy rows valid. Migration 0048 is the expand phase for future Runtime outcome reconciliation readers and storage; this compatibility release does not write its new -values, so a clean 0048-to-0047 downgrade remains supported before writer activation. Once a later -release writes `RECONCILED` observation evidence or `RECONCILE_RUNTIME_*` TaskResolution actions, +values, so a clean 0048-to-0047 downgrade remains supported before writer activation. Once the +writer release stores `RECONCILED` observation evidence or `RECONCILE_RUNTIME_*` TaskResolution actions, 0048 becomes the schema floor. Roll application binaries back to the 0048 compatibility release, not to an older reader, and never rewrite reconciliation audit evidence to force a downgrade. diff --git a/docs/roadmap.md b/docs/roadmap.md index d8406fc..7cbc4f0 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -1,7 +1,7 @@ # Design and delivery roadmap Status: Alpha -Last updated: 2026-08-20 +Last updated: 2026-08-23 路线图使用可验证的垂直切片推进。阶段编号描述交付成熟度，不等同于架构文档的 L0–L3。 各正式 L2 模块的当前代码成熟度与下一交付队列见 @@ -172,6 +172,8 @@ Exit signal：用户可从模板创建公司、绑定真实 Agent，在不伪造 - [x] A4.1b.1 CI/test-only managed DIRECT Worker authority（fenced dispatch、原子终结、 reconciliation-required 停车；不含 reconcile command） - [x] A4.1b.2a reconciliation reader/schema compatibility（0048 expand-only；不含 writer/API） +- [x] A4.1b.2b evidence-driven privileged outcome reconciliation（canonical observation、原子收敛、 + 幂等/并发保护、无 provider redispatch；默认关闭） - [ ] MCP write 和 fake external action 通过统一 Intent/Permit/Receipt/Reconciliation - [ ] Chaos smoke 证明核心 crash windows 收敛且无重复不可逆副作用 @@ -179,7 +181,7 @@ Exit signal：同一部署管理 LangGraph 与非 LangGraph Agent；两者使用 身份、治理、Artifact 和恢复语义，并由机器可读故障报告证明关键不变量。 当前 A4.0 conformance harness 已在 PR #150 完成，A4.1a admission 与 A4.1b.1 managed DIRECT -Worker authority/atomic parking 已交付，A4.1b.2a reader/schema compatibility 已完成。 -A4.1b.2b 仍需受权限控制、证据驱动的 reconcile command； +Worker authority/atomic parking 已交付，A4.1b.2a reader/schema compatibility 与 A4.1b.2b +受权限控制、证据驱动的 reconcile command 已完成； 完整 A4 还需 chaos、parity、reviewed/coordinated cutover 和生产 durable runtime。#135/#136 继续保持开放。 diff --git a/tests/integration/test_managed_direct_worker_postgres.py b/tests/integration/test_managed_direct_worker_postgres.py index a7ea8d1..ec391b5 100644 --- a/tests/integration/test_managed_direct_worker_postgres.py +++ b/tests/integration/test_managed_direct_worker_postgres.py @@ -3,6 +3,7 @@ from __future__ import annotations import os +from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timedelta, timezone from uuid import UUID, uuid4 @@ -12,25 +13,40 @@ from agentmesh.application.managed_runtime_execution import ManagedRuntimeExecutionService from agentmesh.application.quota_services import QuotaPolicyService +from agentmesh.application.runtime_reconciliation import ( + RuntimeOutcomeReconciliationService, +) 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.errors import ( + IdempotencyConflict, + InvalidTaskTransition, + RunLeaseUnavailable, +) +from agentmesh.domain.identity import PrincipalContext, PrincipalType, Role from agentmesh.domain.messaging import MessageEnvelope from agentmesh.domain.quotas import QuotaScope from agentmesh.domain.runtime_execution import ( + RuntimeExecutionPhase, + RuntimeLifecycleIntent, + RuntimeLifecycleOperation, + RuntimeLifecycleStatus, + RuntimeObservationEvidence, RuntimeObservationOutcome, ) from agentmesh.domain.tasks import AttemptStatus, RunStatus, TaskStatus from agentmesh.features import FeatureGateSet from agentmesh.infrastructure.postgres.models import ( + IdempotencyRecordModel, InboxMessageRecord, OutboxEventRecord, QuotaReservationRecord, RuntimeExecutionRecord, RuntimeObservationRecord, + TaskResolutionRecord, ) from agentmesh.infrastructure.postgres.uow import SqlAlchemyUnitOfWorkFactory from agentmesh.infrastructure.runtime.langgraph_adapter import ( @@ -38,7 +54,7 @@ EphemeralRuntimeStateStore, LangGraphManagedAgentRuntime, ) -from agentmesh.runtime_sdk import RuntimeObservation, RuntimePhase +from agentmesh.runtime_sdk import RuntimeObservation, RuntimePhase, canonical_digest pytestmark = [ pytest.mark.postgres, @@ -202,6 +218,100 @@ def _cleanup_task_outbox(factory, task_id) -> None: session.commit() +def _operator(tenant_id: str) -> PrincipalContext: + return PrincipalContext( + principal_id=f"operator-{uuid4().hex}", + tenant_id=tenant_id, + principal_type=PrincipalType.USER, + roles=frozenset({Role.OPERATOR}), + authenticated=True, + authentication_method="postgres-test", + ) + + +def _park_for_reconciliation(*, budget=None): + fixture = _fixture(lease_duration=timedelta(seconds=-1)) + engine, factory, registry, tasks, worker, _backend, _consumer, settings = fixture + task_id, run, envelope = _request( + tasks, settings.tenant_id, factory, budget=budget + ) + 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), + ) + execution = 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 + assert poison.calls == 0 + return (*fixture, task_id, run, attempt, execution, poison) + + +def _confirmed_observation(execution, phase=RuntimePhase.SUCCEEDED, *, observed_at=None): + return RuntimeObservation( + observation_id=str(uuid4()), + runtime_execution_id=str(execution.id), + assignment_id=str(execution.assignment_id), + assignment_digest=execution.assignment_digest, + phase=phase, + observed_at=observed_at or datetime.now(timezone.utc), + provider_event_id=f"postgres-reconcile-{uuid4().hex}", + output={"managed": "reconciled"} if phase is RuntimePhase.SUCCEEDED else None, + ) + + +def _reconciler(factory, settings, **kwargs): + return RuntimeOutcomeReconciliationService( + uow_factory=SqlAlchemyUnitOfWorkFactory(factory), + tenant_id=settings.tenant_id, + feature_gates=_gates(), + **kwargs, + ) + + +class _MemoryProbe: + def __init__(self) -> None: + self.calls = 0 + + def capture_completed_task_in_unit_of_work(self, uow, task): + self.calls += 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("best-effort research failure") + + def test_postgres_managed_authoritative_success_is_atomic_and_replay_safe() -> None: engine, factory, _registry, tasks, worker, backend, consumer, settings = _fixture() task_id = None @@ -440,3 +550,557 @@ def test_postgres_stale_parking_evidence_rolls_back_domain_state() -> None: finally: _cleanup_task_outbox(factory, task_id) engine.dispose() + + +def test_postgres_runtime_outcome_reconciliation_is_atomic_and_replay_safe() -> None: + ( + engine, + factory, + _registry, + tasks, + _worker, + _backend, + _consumer, + settings, + task_id, + run, + attempt, + execution, + poison, + ) = _park_for_reconciliation() + try: + observation = _confirmed_observation(execution) + digest = canonical_digest(observation.to_dict()) + memory = _MemoryProbe() + research = _ResearchProbe(fail=True) + service = _reconciler( + factory, + settings, + runtime_memory_service=memory, + research_materialization_service=research, + ) + principal = _operator(settings.tenant_id) + + first = service.reconcile_outcome( + execution.id, + principal=principal, + observation=observation, + evidence_digest=digest, + evidence_reference="case://postgres/runtime-success", + reason="Provider support confirmed success", + idempotency_key="pg-runtime-reconcile-success", + ) + replay = service.reconcile_outcome( + execution.id, + principal=principal, + observation=observation, + evidence_digest=digest, + evidence_reference="case://postgres/runtime-success", + reason="Provider support confirmed success", + idempotency_key="pg-runtime-reconcile-success", + ) + + aggregate = tasks.get_task(task_id) + assert first.resolution.id == replay.resolution.id + assert aggregate.task.status is TaskStatus.COMPLETED + assert aggregate.task.output == {"managed": "reconciled"} + assert aggregate.runs[0].status is RunStatus.SUCCEEDED + assert aggregate.attempts[0].status is AttemptStatus.SUCCEEDED + assert poison.calls == 0 + assert memory.calls == 1 + assert research.calls == 1 + with factory() as session: + execution_row = session.get(RuntimeExecutionRecord, execution.id) + assert execution_row is not None and execution_row.phase == "SUCCEEDED" + assert session.scalar( + select(func.count()).select_from(RuntimeObservationRecord).where( + RuntimeObservationRecord.runtime_execution_id == execution.id, + RuntimeObservationRecord.processing_outcome == "RECONCILED", + ) + ) == 1 + assert session.scalar( + select(func.count()).select_from(TaskResolutionRecord).where( + TaskResolutionRecord.task_id == task_id + ) + ) == 1 + assert session.scalar( + select(func.count()).select_from(OutboxEventRecord).where( + OutboxEventRecord.envelope["schema_name"].astext + == "agentmesh.runtime.outcome-reconciled", + OutboxEventRecord.envelope["payload"]["run_id"].astext + == str(run.id), + ) + ) == 1 + assert session.scalar( + select(func.count()).select_from(IdempotencyRecordModel).where( + IdempotencyRecordModel.key == "pg-runtime-reconcile-success" + ) + ) == 1 + conflicting = _confirmed_observation(execution, phase=RuntimePhase.FAILED) + with pytest.raises(IdempotencyConflict): + service.reconcile_outcome( + execution.id, + principal=principal, + observation=conflicting, + evidence_digest=canonical_digest(conflicting.to_dict()), + evidence_reference="case://postgres/runtime-failure", + reason="Conflicting conclusion", + idempotency_key="pg-runtime-reconcile-success", + ) + finally: + _cleanup_task_outbox(factory, task_id) + engine.dispose() + + +class _FailingMemoryCapture: + def capture_completed_task_in_unit_of_work(self, uow, task): + raise RuntimeError("memory capture fault") + + +def test_postgres_reconciliation_memory_failure_rolls_back_everything() -> None: + ( + engine, + factory, + _registry, + tasks, + _worker, + _backend, + _consumer, + settings, + task_id, + run, + _attempt, + execution, + poison, + ) = _park_for_reconciliation() + try: + observation = _confirmed_observation(execution) + with pytest.raises(RuntimeError, match="memory capture fault"): + _reconciler( + factory, settings, runtime_memory_service=_FailingMemoryCapture() + ).reconcile_outcome( + execution.id, + principal=_operator(settings.tenant_id), + observation=observation, + evidence_digest=canonical_digest(observation.to_dict()), + evidence_reference="case://postgres/rollback", + reason="Confirmed outcome", + idempotency_key="pg-runtime-reconcile-rollback", + ) + aggregate = tasks.get_task(task_id) + assert aggregate.task.status is TaskStatus.RECONCILIATION_REQUIRED + assert aggregate.runs[0].status is RunStatus.RECONCILIATION_REQUIRED + assert aggregate.attempts[0].status is AttemptStatus.OUTCOME_UNKNOWN + 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(RuntimeObservationRecord).where( + RuntimeObservationRecord.runtime_execution_id == execution.id, + RuntimeObservationRecord.processing_outcome == "RECONCILED", + ) + ) == 0 + assert session.scalar( + select(func.count()).select_from(TaskResolutionRecord).where( + TaskResolutionRecord.task_id == task_id + ) + ) == 0 + assert session.scalar( + select(func.count()).select_from(IdempotencyRecordModel).where( + IdempotencyRecordModel.key == "pg-runtime-reconcile-rollback" + ) + ) == 0 + assert session.scalar( + select(func.count()).select_from(OutboxEventRecord).where( + OutboxEventRecord.envelope["schema_name"].astext + == "agentmesh.runtime.outcome-reconciled", + OutboxEventRecord.envelope["payload"]["run_id"].astext + == str(run.id), + ) + ) == 0 + finally: + _cleanup_task_outbox(factory, task_id) + engine.dispose() + + +def test_postgres_competing_reconciliation_conclusions_have_one_winner() -> None: + ( + engine, + factory, + _registry, + tasks, + _worker, + _backend, + _consumer, + settings, + task_id, + run, + _attempt, + execution, + poison, + ) = _park_for_reconciliation() + try: + success = _confirmed_observation(execution) + failure = _confirmed_observation(execution, phase=RuntimePhase.FAILED) + + def reconcile(observation, key): + return _reconciler(factory, settings).reconcile_outcome( + execution.id, + principal=_operator(settings.tenant_id), + observation=observation, + evidence_digest=canonical_digest(observation.to_dict()), + evidence_reference=f"case://postgres/{key}", + reason="Independent operator conclusion", + idempotency_key=key, + ) + + with ThreadPoolExecutor(max_workers=2) as pool: + futures = [ + pool.submit(reconcile, success, "concurrent-success"), + pool.submit(reconcile, failure, "concurrent-failure"), + ] + results = [] + errors = [] + for future in futures: + try: + results.append(future.result()) + except InvalidTaskTransition as exc: + errors.append(exc) + assert len(results) == len(errors) == 1 + assert poison.calls == 0 + aggregate = tasks.get_task(task_id) + assert aggregate.task.status in {TaskStatus.COMPLETED, TaskStatus.FAILED} + with factory() as session: + assert session.scalar( + select(func.count()).select_from(TaskResolutionRecord).where( + TaskResolutionRecord.task_id == task_id + ) + ) == 1 + assert session.scalar( + select(func.count()).select_from(RuntimeObservationRecord).where( + RuntimeObservationRecord.runtime_execution_id == execution.id, + RuntimeObservationRecord.processing_outcome == "RECONCILED", + ) + ) == 1 + finally: + _cleanup_task_outbox(factory, task_id) + engine.dispose() + + +@pytest.mark.parametrize( + ("phase", "expected_task", "expected_run", "expected_attempt", "reason"), + [ + ( + RuntimePhase.FAILED, + TaskStatus.FAILED, + RunStatus.FAILED, + AttemptStatus.FAILED, + "runtime.reconciled_failed", + ), + ( + RuntimePhase.TIMED_OUT, + TaskStatus.FAILED, + RunStatus.FAILED, + AttemptStatus.FAILED, + "runtime.reconciled_timed_out", + ), + ( + RuntimePhase.CANCELED, + TaskStatus.FAILED, + RunStatus.FAILED, + AttemptStatus.FAILED, + "runtime.unrequested_cancellation", + ), + ], +) +def test_postgres_reconciliation_known_non_success_mapping( + phase, expected_task, expected_run, expected_attempt, reason +) -> None: + ( + engine, + factory, + _registry, + tasks, + _worker, + _backend, + _consumer, + settings, + task_id, + _run, + _attempt, + execution, + poison, + ) = _park_for_reconciliation() + try: + observation = _confirmed_observation(execution, phase=phase) + _reconciler(factory, settings).reconcile_outcome( + execution.id, + principal=_operator(settings.tenant_id), + observation=observation, + evidence_digest=canonical_digest(observation.to_dict()), + evidence_reference=f"case://postgres/{phase.value.lower()}", + reason="Confirmed terminal outcome", + idempotency_key=f"known-{phase.value.lower()}-{uuid4().hex}", + ) + aggregate = tasks.get_task(task_id) + assert aggregate.task.status is expected_task + assert aggregate.task.error == reason + assert aggregate.runs[0].status is expected_run + assert aggregate.attempts[0].status is expected_attempt + assert poison.calls == 0 + finally: + _cleanup_task_outbox(factory, task_id) + engine.dispose() + + +def test_postgres_requested_cancellation_maps_all_business_state_to_canceled() -> None: + ( + engine, + factory, + _registry, + tasks, + _worker, + _backend, + _consumer, + settings, + task_id, + _run, + _attempt, + execution, + poison, + ) = _park_for_reconciliation() + try: + now = datetime.now(timezone.utc) + with SqlAlchemyUnitOfWorkFactory(factory)() as uow: + uow.runtimes.add_lifecycle_operation( + RuntimeLifecycleIntent( + id=uuid4(), + tenant_id=settings.tenant_id, + runtime_execution_id=execution.id, + operation_id=f"operator-cancel-{uuid4().hex}", + operation=RuntimeLifecycleOperation.CANCEL, + intent_digest="f" * 64, + status=RuntimeLifecycleStatus.REQUESTED, + deadline=now + timedelta(minutes=10), + receipt_summary=None, + version=1, + created_at=now, + updated_at=now, + ) + ) + uow.commit() + observation = _confirmed_observation(execution, phase=RuntimePhase.CANCELED) + result = _reconciler(factory, settings).reconcile_outcome( + execution.id, + principal=_operator(settings.tenant_id), + observation=observation, + evidence_digest=canonical_digest(observation.to_dict()), + evidence_reference="case://postgres/requested-cancel", + reason="Provider confirmed requested cancellation", + idempotency_key=f"requested-cancel-{uuid4().hex}", + ) + aggregate = tasks.get_task(task_id) + assert aggregate.task.status is TaskStatus.CANCELED + assert aggregate.runs[0].status is RunStatus.CANCELED + assert aggregate.attempts[0].status is AttemptStatus.CANCELED + assert result.resolution.details["business_mapping_reason"] == ( + "runtime.reconciled_canceled" + ) + assert poison.calls == 0 + finally: + _cleanup_task_outbox(factory, task_id) + engine.dispose() + + +def test_postgres_success_at_budget_deadline_waits_for_approval_without_resettling() -> None: + deadline = datetime.now(timezone.utc) + timedelta(minutes=10) + budget = TaskBudget.create(deadline=deadline) + ( + engine, + factory, + _registry, + tasks, + _worker, + _backend, + _consumer, + settings, + task_id, + _run, + attempt, + execution, + poison, + ) = _park_for_reconciliation(budget=budget) + try: + parked = tasks.get_task(task_id) + settlement_source = parked.attempts[0].budget_settlement_source + settled_tokens = parked.task.settled_tokens + observation = _confirmed_observation(execution, observed_at=deadline) + result = _reconciler(factory, settings).reconcile_outcome( + execution.id, + principal=_operator(settings.tenant_id), + observation=observation, + evidence_digest=canonical_digest(observation.to_dict()), + evidence_reference="case://postgres/deadline", + reason="Success confirmed at the pinned deadline", + idempotency_key=f"deadline-{uuid4().hex}", + ) + aggregate = tasks.get_task(task_id) + assert aggregate.task.status is TaskStatus.WAITING_APPROVAL + assert aggregate.task.current_run_id is None + assert aggregate.task.candidate_output == {"managed": "reconciled"} + assert aggregate.task.budget_exhausted_reason == "budget_deadline_exceeded" + assert aggregate.runs[0].status is RunStatus.SUCCEEDED + assert aggregate.attempts[0].status is AttemptStatus.SUCCEEDED + assert aggregate.attempts[0].budget_settlement_source is settlement_source + assert aggregate.task.settled_tokens == settled_tokens + assert result.resolution.resulting_status is TaskStatus.WAITING_APPROVAL + assert result.resolution.details["business_mapping_reason"] == ( + "budget_deadline_exceeded" + ) + assert aggregate.attempts[0].id == attempt.id + assert poison.calls == 0 + finally: + _cleanup_task_outbox(factory, task_id) + engine.dispose() + + +def test_postgres_reuses_exact_late_evidence_without_duplicate_record() -> None: + ( + engine, + factory, + _registry, + tasks, + _worker, + _backend, + _consumer, + settings, + task_id, + _run, + _attempt, + execution, + poison, + ) = _park_for_reconciliation() + try: + observation = _confirmed_observation(execution) + digest = canonical_digest(observation.to_dict()) + evidence_id = uuid4() + with SqlAlchemyUnitOfWorkFactory(factory)() as uow: + uow.runtimes.add_observation( + RuntimeObservationEvidence( + id=evidence_id, + tenant_id=settings.tenant_id, + runtime_execution_id=execution.id, + observation_id=observation.observation_id, + observation_digest=digest, + assignment_id=execution.assignment_id, + assignment_digest=execution.assignment_digest, + provider_sequence=observation.provider_sequence, + phase=RuntimeExecutionPhase.SUCCEEDED, + observed_at=observation.observed_at, + received_at=datetime.now(timezone.utc), + safe_summary="Late terminal evidence", + processing_outcome=RuntimeObservationOutcome.CONFLICT, + provider_event_present=True, + evidence={ + "provider_event_id": observation.provider_event_id, + "snapshot_digest": observation.snapshot_digest, + }, + ) + ) + uow.commit() + _reconciler(factory, settings).reconcile_outcome( + execution.id, + principal=_operator(settings.tenant_id), + observation=observation, + evidence_digest=digest, + evidence_reference="case://postgres/existing-evidence", + reason="Existing evidence independently verified", + idempotency_key=f"existing-evidence-{uuid4().hex}", + ) + with factory() as session: + records = list( + session.scalars( + select(RuntimeObservationRecord).where( + RuntimeObservationRecord.runtime_execution_id == execution.id, + RuntimeObservationRecord.observation_id + == observation.observation_id, + ) + ) + ) + assert len(records) == 1 + assert records[0].id == evidence_id + assert records[0].processing_outcome == "RECONCILED" + assert tasks.get_task(task_id).task.status is TaskStatus.COMPLETED + assert poison.calls == 0 + finally: + _cleanup_task_outbox(factory, task_id) + engine.dispose() + + +def test_postgres_stale_reconciliation_fence_has_zero_side_effects() -> None: + ( + engine, + factory, + _registry, + tasks, + _worker, + _backend, + _consumer, + settings, + task_id, + run, + _attempt, + execution, + poison, + ) = _park_for_reconciliation() + try: + with factory() as session: + row = session.get(RuntimeExecutionRecord, execution.id) + assert row is not None and row.current_fencing_token is not None + row.current_fencing_token += 1 + session.commit() + observation = _confirmed_observation(execution) + with pytest.raises(InvalidTaskTransition, match="strictly consistent"): + _reconciler(factory, settings).reconcile_outcome( + execution.id, + principal=_operator(settings.tenant_id), + observation=observation, + evidence_digest=canonical_digest(observation.to_dict()), + evidence_reference="case://postgres/stale-fence", + reason="Stale evidence must fail", + idempotency_key="stale-fence-reconciliation", + ) + aggregate = tasks.get_task(task_id) + assert aggregate.task.status is TaskStatus.RECONCILIATION_REQUIRED + assert aggregate.runs[0].status is RunStatus.RECONCILIATION_REQUIRED + assert aggregate.attempts[0].status is AttemptStatus.OUTCOME_UNKNOWN + assert poison.calls == 0 + with factory() as session: + assert session.scalar( + select(func.count()).select_from(RuntimeObservationRecord).where( + RuntimeObservationRecord.runtime_execution_id == execution.id, + RuntimeObservationRecord.processing_outcome == "RECONCILED", + ) + ) == 0 + assert session.scalar( + select(func.count()).select_from(TaskResolutionRecord).where( + TaskResolutionRecord.task_id == task_id + ) + ) == 0 + assert session.scalar( + select(func.count()).select_from(IdempotencyRecordModel).where( + IdempotencyRecordModel.key == "stale-fence-reconciliation" + ) + ) == 0 + assert session.scalar( + select(func.count()).select_from(OutboxEventRecord).where( + OutboxEventRecord.envelope["schema_name"].astext + == "agentmesh.runtime.outcome-reconciled", + OutboxEventRecord.envelope["payload"]["run_id"].astext + == str(run.id), + ) + ) == 0 + finally: + _cleanup_task_outbox(factory, task_id) + engine.dispose() diff --git a/tests/test_runtime_reconciliation_service.py b/tests/test_runtime_reconciliation_service.py new file mode 100644 index 0000000..5fd306c --- /dev/null +++ b/tests/test_runtime_reconciliation_service.py @@ -0,0 +1,112 @@ +from datetime import datetime, timezone +from uuid import uuid4 + +import pytest + +from agentmesh.application.runtime_reconciliation import RuntimeOutcomeReconciliationService +from agentmesh.domain.errors import AuthorizationDenied, InvalidTaskInput +from agentmesh.domain.identity import PrincipalContext, PrincipalType, Role +from agentmesh.features import FeatureGateSet +from agentmesh.runtime_sdk import RuntimeObservation, RuntimePhase, canonical_digest + +TENANT_ID = "runtime-reconciliation-unit" + + +def _principal(*, tenant_id: str = TENANT_ID, authenticated: bool = True) -> PrincipalContext: + return PrincipalContext( + principal_id="operator-unit", + tenant_id=tenant_id, + principal_type=PrincipalType.USER, + roles=frozenset({Role.OPERATOR}), + authenticated=authenticated, + authentication_method="test", + ) + + +def _service() -> RuntimeOutcomeReconciliationService: + def reject_uow(): + raise AssertionError("invalid requests must be rejected before opening a UoW") + + return RuntimeOutcomeReconciliationService( + uow_factory=reject_uow, + tenant_id=TENANT_ID, + feature_gates=FeatureGateSet.from_config( + "full", + "managed_agent_runtime=true,outcome_reconciliation=true,identity_rbac=true", + ), + ) + + +def _observation( + execution_id, + *, + phase: RuntimePhase = RuntimePhase.SUCCEEDED, + output=None, + usage=None, +) -> RuntimeObservation: + return RuntimeObservation( + observation_id=str(uuid4()), + runtime_execution_id=str(execution_id), + assignment_id=str(uuid4()), + assignment_digest="a" * 64, + phase=phase, + observed_at=datetime.now(timezone.utc), + snapshot_digest="b" * 64, + output={"answer": 42} if output is None and phase is RuntimePhase.SUCCEEDED else output, + usage={} if usage is None else usage, + ) + + +def _reconcile(service, execution_id, observation, **overrides): + values = { + "principal": _principal(), + "observation": observation, + "evidence_digest": canonical_digest(observation.to_dict()), + "evidence_reference": "case://unit/evidence", + "reason": "Provider support supplied canonical evidence", + "idempotency_key": "unit-reconcile-1", + } + values.update(overrides) + return service.reconcile_outcome(execution_id, **values) + + +@pytest.mark.parametrize( + ("principal", "expected"), + [ + (_principal(authenticated=False), AuthorizationDenied), + (_principal(tenant_id="another-tenant"), AuthorizationDenied), + ], +) +def test_runtime_reconciliation_rejects_invalid_principal_before_uow(principal, expected) -> None: + execution_id = uuid4() + observation = _observation(execution_id) + with pytest.raises(expected): + _reconcile(_service(), execution_id, observation, principal=principal) + + +def test_runtime_reconciliation_rejects_nonterminal_and_wrong_execution_before_uow() -> None: + service = _service() + execution_id = uuid4() + running = _observation(execution_id, phase=RuntimePhase.RUNNING) + with pytest.raises(InvalidTaskInput, match="known terminal"): + _reconcile(service, execution_id, running) + + other_execution = _observation(uuid4()) + with pytest.raises(InvalidTaskInput, match="execution identity"): + _reconcile(service, execution_id, other_execution) + + +def test_runtime_reconciliation_rejects_digest_and_success_shape_before_uow() -> None: + service = _service() + execution_id = uuid4() + observation = _observation(execution_id) + with pytest.raises(InvalidTaskInput, match="canonical observation digest"): + _reconcile(service, execution_id, observation, evidence_digest="f" * 64) + + non_mapping = _observation(execution_id, output=["not", "a", "mapping"]) + with pytest.raises(InvalidTaskInput, match="mapping output"): + _reconcile(service, execution_id, non_mapping) + + billed = _observation(execution_id, usage={"input_tokens": 1}) + with pytest.raises(InvalidTaskInput, match="empty usage"): + _reconcile(service, execution_id, billed) From e297f8820177cf1cb5488c2b9141bc36c91c1f60 Mon Sep 17 00:00:00 2001 From: 0YHR0 <97868579@qq.com> Date: Mon, 24 Aug 2026 10:36:01 +0800 Subject: [PATCH 4/7] fix(runtime): reject unsafe reconciliation evidence --- .../modules/runtime-outcome-reconciliation.md | 9 ++- docs/implementation-status.md | 4 +- src/agentmesh/api/runtime_routes.py | 19 ++++- .../application/runtime_reconciliation.py | 19 ++++- .../test_managed_direct_worker_postgres.py | 33 ++++++++- tests/test_runtime_execution_domain.py | 8 +- tests/test_runtime_reconciliation_service.py | 74 ++++++++++++++++++- tests/test_runtime_routes.py | 57 ++++++++++++++ 8 files changed, 210 insertions(+), 13 deletions(-) diff --git a/docs/architecture/modules/runtime-outcome-reconciliation.md b/docs/architecture/modules/runtime-outcome-reconciliation.md index 6ee5d51..158701b 100644 --- a/docs/architecture/modules/runtime-outcome-reconciliation.md +++ b/docs/architecture/modules/runtime-outcome-reconciliation.md @@ -14,8 +14,13 @@ work. The request contains a complete public `RuntimeObservation`, its canonical digest, a bounded evidence reference, and a bounded operator reason. Only `SUCCEEDED`, `FAILED`, `CANCELED`, and `TIMED_OUT` observations are accepted. Execution, assignment, digest, phase, and provider identity -evidence must match the persisted Runtime execution. Success remains limited to mapping output and -empty usage; other terminal phases cannot carry successful output. +evidence must match the persisted Runtime execution. All phases require empty usage because parking +has already conservatively settled budget and this slice does not yet support actual-usage evidence. +Terminal evidence cannot retain governed-action or wait requests, and success cannot carry an error. +Success remains limited to mapping output; `output_artifact_refs` may accompany the canonical +observation but this slice terminates the Task only from the mapping output. Artifact materialization +and post-reconciliation actual-usage accounting are explicit follow-up work. Other terminal phases +cannot carry successful output. ## Atomic convergence diff --git a/docs/implementation-status.md b/docs/implementation-status.md index 4cbc01b..371da12 100644 --- a/docs/implementation-status.md +++ b/docs/implementation-status.md @@ -160,7 +160,9 @@ A4.1b.2b evidence-driven outcome reconciliation (writer slice): revalidates the Task/Run/latest Attempt/Runtime quartet, never calls or redispatches a provider, and commits evidence, Runtime and business convergence, TaskResolution, Outbox, and idempotency in one UoW. Exact replay is side-effect free; conflicting evidence fails closed. -- Confirmed success retains the A4.1 mapping-output/empty-usage limit. An observation at or after +- All confirmed outcomes require empty usage because parking has already conservatively settled + budget; terminal evidence also rejects remaining governed-action/wait requests, and success + rejects an error. Confirmed success retains the A4.1 mapping-output limit. An observation at or after the pinned deadline preserves the provider success but moves the Task to `WAITING_APPROVAL` with candidate output. Confirmed cancellation maps business objects to `CANCELED` only when a persisted cancel intent exists; otherwise they fail as `runtime.unrequested_cancellation`. diff --git a/src/agentmesh/api/runtime_routes.py b/src/agentmesh/api/runtime_routes.py index 3f2a4bd..f8235ce 100644 --- a/src/agentmesh/api/runtime_routes.py +++ b/src/agentmesh/api/runtime_routes.py @@ -15,7 +15,7 @@ RuntimeOutcomeReconciliationService, ) from agentmesh.application.runtime_services import RuntimeRegistryService -from agentmesh.domain.errors import AuthorizationDenied +from agentmesh.domain.errors import AuthorizationDenied, InvalidTaskInput from agentmesh.domain.identity import Permission from agentmesh.domain.runtime_execution import ( RuntimeExecution, @@ -23,7 +23,7 @@ RuntimeVersion, ) from agentmesh.features import Feature -from agentmesh.runtime_sdk import RuntimeObservation +from agentmesh.runtime_sdk import RuntimeContractError, RuntimeObservation router = APIRouter(prefix="/api/v1", tags=["runtime-control-plane"]) _dependencies = [ @@ -97,7 +97,10 @@ class RuntimeObservationResponse(BaseModel): class ReconcileRuntimeOutcomeRequest(BaseModel): model_config = ConfigDict(extra="forbid") - observation: dict[str, Any] + # Keep the versioned Runtime contract opaque to Pydantic so malformed + # values are normalized by the SDK boundary below instead of being echoed + # in FastAPI's default validation response. + observation: Any evidence_digest: str evidence_reference: str reason: str @@ -260,7 +263,15 @@ def reconcile_runtime_outcome( ) -> ReconcileRuntimeOutcomeResponse: if principal.tenant_id != service.tenant_id or not principal.authenticated: raise AuthorizationDenied("Runtime tenant scope denied") - observation = RuntimeObservation.from_dict(payload.observation) + try: + observation = RuntimeObservation.from_dict(payload.observation) + except RuntimeContractError as exc: + # Runtime contract errors can contain field-level details derived from an + # untrusted request. Keep the public error stable and bounded while the + # domain exception handler maps it to HTTP 422. + raise InvalidTaskInput( + "Runtime reconciliation observation is invalid" + ) from exc result: RuntimeOutcomeReconciliationResult = service.reconcile_outcome( execution_id, principal=principal, diff --git a/src/agentmesh/application/runtime_reconciliation.py b/src/agentmesh/application/runtime_reconciliation.py index a25edcb..b53cfb2 100644 --- a/src/agentmesh/application/runtime_reconciliation.py +++ b/src/agentmesh/application/runtime_reconciliation.py @@ -94,16 +94,33 @@ def reconcile_outcome( raise IdempotencyConflict("Idempotency-Key must not be empty") if observation.phase not in _KNOWN_TERMINAL_PHASES: raise InvalidTaskInput("Reconciliation requires a known terminal observation") + if observation.usage: + raise InvalidTaskInput( + "Runtime reconciliation requires empty usage until usage evidence is supported" + ) + if observation.governed_action_requests or observation.wait_refs: + raise InvalidTaskInput( + "Terminal Runtime evidence cannot retain action or wait requests" + ) + if ( + observation.provider_event_id is not None + and len(observation.provider_event_id.encode("utf-8")) > 512 + ): + raise InvalidTaskInput( + "Runtime reconciliation provider event identity exceeds the persistence limit" + ) observation_digest = canonical_digest(observation.to_dict()) if evidence_digest != observation_digest: raise InvalidTaskInput("Evidence digest must equal the canonical observation digest") if UUID(observation.runtime_execution_id) != execution_id: raise InvalidTaskInput("Observation Runtime execution identity does not match") if observation.phase is RuntimePhase.SUCCEEDED: - if type(observation.output) is not dict or observation.usage: + if type(observation.output) is not dict: raise InvalidTaskInput( "Managed Runtime success requires mapping output and empty usage" ) + if observation.error is not None: + raise InvalidTaskInput("Runtime success evidence cannot carry an error") elif observation.output is not None or observation.output_artifact_refs: raise InvalidTaskInput("Non-success Runtime evidence cannot carry output") diff --git a/tests/integration/test_managed_direct_worker_postgres.py b/tests/integration/test_managed_direct_worker_postgres.py index ec391b5..4de1bfd 100644 --- a/tests/integration/test_managed_direct_worker_postgres.py +++ b/tests/integration/test_managed_direct_worker_postgres.py @@ -229,9 +229,19 @@ def _operator(tenant_id: str) -> PrincipalContext: ) -def _park_for_reconciliation(*, budget=None): +def _park_for_reconciliation(*, budget=None, quota: bool = False): fixture = _fixture(lease_duration=timedelta(seconds=-1)) engine, factory, registry, tasks, worker, _backend, _consumer, settings = fixture + if quota: + QuotaPolicyService( + SqlAlchemyUnitOfWorkFactory(factory), settings.tenant_id + ).put_policy( + scope=QuotaScope.TENANT, + project_id=None, + max_concurrent_attempts=1, + weight=1, + created_by="postgres-reconciliation-test", + ) task_id, run, envelope = _request( tasks, settings.tenant_id, factory, budget=budget ) @@ -930,11 +940,20 @@ def test_postgres_success_at_budget_deadline_waits_for_approval_without_resettli attempt, execution, poison, - ) = _park_for_reconciliation(budget=budget) + ) = _park_for_reconciliation(budget=budget, quota=True) try: parked = tasks.get_task(task_id) settlement_source = parked.attempts[0].budget_settlement_source settled_tokens = parked.task.settled_tokens + with factory() as session: + reservation_before = session.scalar( + select(QuotaReservationRecord).where( + QuotaReservationRecord.attempt_id == attempt.id + ) + ) + assert reservation_before is not None + released_at = reservation_before.released_at + assert released_at is not None observation = _confirmed_observation(execution, observed_at=deadline) result = _reconciler(factory, settings).reconcile_outcome( execution.id, @@ -954,6 +973,16 @@ def test_postgres_success_at_budget_deadline_waits_for_approval_without_resettli assert aggregate.attempts[0].status is AttemptStatus.SUCCEEDED assert aggregate.attempts[0].budget_settlement_source is settlement_source assert aggregate.task.settled_tokens == settled_tokens + with factory() as session: + reservations = list( + session.scalars( + select(QuotaReservationRecord).where( + QuotaReservationRecord.attempt_id == attempt.id + ) + ) + ) + assert len(reservations) == 1 + assert reservations[0].released_at == released_at assert result.resolution.resulting_status is TaskStatus.WAITING_APPROVAL assert result.resolution.details["business_mapping_reason"] == ( "budget_deadline_exceeded" diff --git a/tests/test_runtime_execution_domain.py b/tests/test_runtime_execution_domain.py index f003341..2f87cd2 100644 --- a/tests/test_runtime_execution_domain.py +++ b/tests/test_runtime_execution_domain.py @@ -50,11 +50,15 @@ def test_observation_preserves_sequence_when_provider_omits_it() -> None: RuntimeExecutionPhase.TIMED_OUT, ], ) +@pytest.mark.parametrize( + "ambiguous_phase", + [RuntimeExecutionPhase.OUTCOME_UNKNOWN, RuntimeExecutionPhase.LOST], +) def test_ambiguous_execution_has_a_dedicated_terminal_reconciliation_exit( - phase: RuntimeExecutionPhase, + phase: RuntimeExecutionPhase, ambiguous_phase: RuntimeExecutionPhase ) -> None: ambiguous = _execution().apply_observation( - phase=RuntimeExecutionPhase.OUTCOME_UNKNOWN, provider_sequence=1 + phase=ambiguous_phase, provider_sequence=1 ) reconciled = ambiguous.reconcile_terminal(phase=phase, provider_sequence=2) diff --git a/tests/test_runtime_reconciliation_service.py b/tests/test_runtime_reconciliation_service.py index 5fd306c..e7bc368 100644 --- a/tests/test_runtime_reconciliation_service.py +++ b/tests/test_runtime_reconciliation_service.py @@ -7,7 +7,14 @@ from agentmesh.domain.errors import AuthorizationDenied, InvalidTaskInput from agentmesh.domain.identity import PrincipalContext, PrincipalType, Role from agentmesh.features import FeatureGateSet -from agentmesh.runtime_sdk import RuntimeObservation, RuntimePhase, canonical_digest +from agentmesh.runtime_sdk import ( + ErrorCategory, + RetryDisposition, + RuntimeErrorDTO, + RuntimeObservation, + RuntimePhase, + canonical_digest, +) TENANT_ID = "runtime-reconciliation-unit" @@ -110,3 +117,68 @@ def test_runtime_reconciliation_rejects_digest_and_success_shape_before_uow() -> billed = _observation(execution_id, usage={"input_tokens": 1}) with pytest.raises(InvalidTaskInput, match="empty usage"): _reconcile(service, execution_id, billed) + + +@pytest.mark.parametrize( + "phase", + [ + RuntimePhase.SUCCEEDED, + RuntimePhase.FAILED, + RuntimePhase.CANCELED, + RuntimePhase.TIMED_OUT, + ], +) +def test_runtime_reconciliation_rejects_usage_for_every_terminal_phase(phase) -> None: + service = _service() + execution_id = uuid4() + observation = _observation(execution_id, phase=phase, usage={"input_tokens": 1}) + with pytest.raises(InvalidTaskInput, match="empty usage"): + _reconcile(service, execution_id, observation) + + +@pytest.mark.parametrize( + "field", + ["governed_action_requests", "wait_refs"], +) +def test_runtime_reconciliation_rejects_unresolved_terminal_requests(field) -> None: + service = _service() + execution_id = uuid4() + values = { + "governed_action_requests": ({"action": "pending"},), + "wait_refs": ("wait://pending",), + } + observation = RuntimeObservation( + observation_id=str(uuid4()), + runtime_execution_id=str(execution_id), + assignment_id=str(uuid4()), + assignment_digest="a" * 64, + phase=RuntimePhase.FAILED, + observed_at=datetime.now(timezone.utc), + snapshot_digest="b" * 64, + **{field: values[field]}, + ) + with pytest.raises(InvalidTaskInput, match="action or wait requests"): + _reconcile(service, execution_id, observation) + + +def test_runtime_reconciliation_rejects_success_with_error() -> None: + service = _service() + execution_id = uuid4() + observation = RuntimeObservation( + observation_id=str(uuid4()), + runtime_execution_id=str(execution_id), + assignment_id=str(uuid4()), + assignment_digest="a" * 64, + phase=RuntimePhase.SUCCEEDED, + observed_at=datetime.now(timezone.utc), + snapshot_digest="b" * 64, + output={"answer": 42}, + error=RuntimeErrorDTO( + code="provider.error", + category=ErrorCategory.PERMANENT, + message="must not accompany success", + retry_disposition=RetryDisposition.NEVER, + ), + ) + with pytest.raises(InvalidTaskInput, match="cannot carry an error"): + _reconcile(service, execution_id, observation) diff --git a/tests/test_runtime_routes.py b/tests/test_runtime_routes.py index e75eae3..985db43 100644 --- a/tests/test_runtime_routes.py +++ b/tests/test_runtime_routes.py @@ -366,3 +366,60 @@ def test_runtime_reconciliation_http_rejects_cross_tenant_and_missing_permission assert cross_tenant.status_code == 403 assert denied.status_code == 403 assert service.calls == [] + + +@pytest.mark.parametrize( + "mutate", + [ + lambda observation: ["not", "an", "observation"], + lambda observation: {**observation, "untrusted_field": "do-not-reflect-me"}, + lambda observation: {**observation, "schema_version": 99}, + lambda observation: { + **observation, + "progress": {"untrusted_payload": "x" * 66_000}, + }, + ], + ids=["malformed", "unknown-field", "unknown-major", "oversize"], +) +def test_runtime_reconciliation_http_maps_invalid_contract_to_safe_422( + application_container, mutate +) -> None: + projection = _ProjectionService() + service = _ReconciliationService(projection.execution) + application_container.runtime_service = projection + application_container.runtime_reconciliation_service = service + application_container.feature_gates = FeatureGateSet.from_config( + "full", + "managed_agent_runtime=true,outcome_reconciliation=true,identity_rbac=true", + ) + principal = _principal(service.tenant_id) + observation = RuntimeObservation( + observation_id=str(uuid4()), + runtime_execution_id=str(projection.execution.id), + assignment_id=str(projection.execution.assignment_id), + assignment_digest=projection.execution.assignment_digest, + phase=RuntimePhase.FAILED, + observed_at=datetime.now(timezone.utc), + snapshot_digest="a" * 64, + ) + encoded = jsonable_encoder(observation.to_dict()) + payload = { + "observation": mutate(encoded), + "evidence_digest": "b" * 64, + "evidence_reference": "case://runtime/invalid", + "reason": "Invalid contract must fail safely", + } + application = create_app(application_container) + application.dependency_overrides[get_principal_context] = lambda: principal + with TestClient(application) as client: + response = client.post( + f"/api/v1/runtime-executions/{projection.execution.id}/reconcile-outcome", + json=payload, + headers={"Idempotency-Key": "invalid-contract"}, + ) + assert response.status_code == 422 + assert response.json() == { + "code": "invalid_task_input", + "message": "Runtime reconciliation observation is invalid", + } + assert service.calls == [] From 2b27980570cbfa32285f6a05f58d607e3a1d1ea3 Mon Sep 17 00:00:00 2001 From: 0YHR0 <97868579@qq.com> Date: Mon, 24 Aug 2026 10:42:01 +0800 Subject: [PATCH 5/7] test(runtime): isolate reconciliation quota admission --- .../test_managed_direct_worker_postgres.py | 25 +++++++++++++------ 1 file changed, 17 insertions(+), 8 deletions(-) diff --git a/tests/integration/test_managed_direct_worker_postgres.py b/tests/integration/test_managed_direct_worker_postgres.py index 4de1bfd..595d377 100644 --- a/tests/integration/test_managed_direct_worker_postgres.py +++ b/tests/integration/test_managed_direct_worker_postgres.py @@ -113,15 +113,21 @@ def execute_authoritative(self, *args, **kwargs): raise AssertionError("crossed execution must park without redispatch") -def _gates() -> FeatureGateSet: +def _gates(*, quota_admission: bool = False) -> 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", + "managed_runtime_direct_cutover=true,identity_rbac=true," + f"quota_admission={'true' if quota_admission else 'false'}", ) -def _fixture(*, lease_duration=timedelta(minutes=5), registry_type=RuntimeRegistryService): +def _fixture( + *, + lease_duration=timedelta(minutes=5), + registry_type=RuntimeRegistryService, + quota_admission: bool = False, +): settings = get_settings() engine = create_engine(settings.database_url) factory = sessionmaker(bind=engine, expire_on_commit=False, class_=Session) @@ -142,16 +148,17 @@ def _fixture(*, lease_duration=timedelta(minutes=5), registry_type=RuntimeRegist ) session.commit() uow_factory = SqlAlchemyUnitOfWorkFactory(factory) + gates = _gates(quota_admission=quota_admission) registry = registry_type( uow_factory=uow_factory, tenant_id=settings.tenant_id, - feature_gates=_gates(), + feature_gates=gates, ) tasks = TaskApplicationService( uow_factory=uow_factory, agent_id=settings.agent_id, tenant_id=settings.tenant_id, - feature_gates=_gates(), + feature_gates=gates, runtime_registry_service=registry, ) backend = _DeterministicBackend() @@ -174,7 +181,7 @@ def _fixture(*, lease_duration=timedelta(minutes=5), registry_type=RuntimeRegist worker_id=consumer, consumer_name=consumer, lease_duration=lease_duration, - feature_gates=_gates(), + feature_gates=gates, ) return engine, factory, registry, tasks, worker, backend, consumer, settings @@ -230,7 +237,9 @@ def _operator(tenant_id: str) -> PrincipalContext: def _park_for_reconciliation(*, budget=None, quota: bool = False): - fixture = _fixture(lease_duration=timedelta(seconds=-1)) + fixture = _fixture( + lease_duration=timedelta(seconds=-1), quota_admission=quota + ) engine, factory, registry, tasks, worker, _backend, _consumer, settings = fixture if quota: QuotaPolicyService( @@ -403,7 +412,7 @@ def test_postgres_managed_finalization_fault_rolls_back_all_authority() -> None: def test_postgres_expired_dispatching_owner_parks_atomically_once() -> None: engine, factory, registry, tasks, worker, _backend, consumer, settings = _fixture( - lease_duration=timedelta(seconds=-1) + lease_duration=timedelta(seconds=-1), quota_admission=True ) task_id = None try: From 81d1ae39d2e759ccb46189f9787e7428c611cf75 Mon Sep 17 00:00:00 2001 From: 0YHR0 <97868579@qq.com> Date: Mon, 24 Aug 2026 10:45:01 +0800 Subject: [PATCH 6/7] test(runtime): enable reconciliation writer gate --- tests/integration/test_managed_direct_worker_postgres.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/integration/test_managed_direct_worker_postgres.py b/tests/integration/test_managed_direct_worker_postgres.py index 595d377..e5e9283 100644 --- a/tests/integration/test_managed_direct_worker_postgres.py +++ b/tests/integration/test_managed_direct_worker_postgres.py @@ -117,7 +117,8 @@ def _gates(*, quota_admission: bool = False) -> FeatureGateSet: return FeatureGateSet.from_config( "full", "managed_agent_runtime=true,managed_runtime_worker=true," - "managed_runtime_direct_cutover=true,identity_rbac=true," + "managed_runtime_direct_cutover=true,outcome_reconciliation=true," + "identity_rbac=true," f"quota_admission={'true' if quota_admission else 'false'}", ) From 131a8e056770adfc2ef2479dfd9ce0b5d0ffb71e Mon Sep 17 00:00:00 2001 From: 0YHR0 <97868579@qq.com> Date: Mon, 24 Aug 2026 10:48:09 +0800 Subject: [PATCH 7/7] test(runtime): clean writer-only postgres evidence --- .../test_managed_direct_worker_postgres.py | 26 +++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/tests/integration/test_managed_direct_worker_postgres.py b/tests/integration/test_managed_direct_worker_postgres.py index e5e9283..66fa3a6 100644 --- a/tests/integration/test_managed_direct_worker_postgres.py +++ b/tests/integration/test_managed_direct_worker_postgres.py @@ -47,6 +47,7 @@ RuntimeExecutionRecord, RuntimeObservationRecord, TaskResolutionRecord, + TaskRunRecord, ) from agentmesh.infrastructure.postgres.uow import SqlAlchemyUnitOfWorkFactory from agentmesh.infrastructure.runtime.langgraph_adapter import ( @@ -215,6 +216,31 @@ def _cleanup_task_outbox(factory, task_id) -> None: return expected = str(task_id) with factory() as session: + # Writer tests intentionally persist 0048-only enum values. Remove + # only those rows belonging to this test Task after assertions so the + # shared suite can still exercise the pre-write 0048 -> 0047 downgrade. + execution_ids = select(RuntimeExecutionRecord.id).join( + TaskRunRecord, TaskRunRecord.id == RuntimeExecutionRecord.run_id + ).where(TaskRunRecord.task_id == task_id) + session.execute( + delete(RuntimeObservationRecord).where( + RuntimeObservationRecord.runtime_execution_id.in_(execution_ids), + RuntimeObservationRecord.processing_outcome == "RECONCILED", + ) + ) + session.execute( + delete(TaskResolutionRecord).where( + TaskResolutionRecord.task_id == task_id, + TaskResolutionRecord.action.in_( + [ + "RECONCILE_RUNTIME_SUCCEEDED", + "RECONCILE_RUNTIME_FAILED", + "RECONCILE_RUNTIME_CANCELED", + "RECONCILE_RUNTIME_TIMED_OUT", + ] + ), + ) + ) for record in session.scalars(select(OutboxEventRecord)): envelope = record.envelope payload = envelope.get("payload", {})