feat(runtime): add managed DIRECT worker authority

docs/implementation-status.md

@@ -1,7 +1,7 @@
 # Implementation status
 
 Status: Alpha baseline
-Last updated: 2026-08-20
+Last updated: 2026-08-21
 
 This page records what the repository actually implements. The formal L2 documents describe the
 target architecture; an implemented vertical slice does not imply that every capability in its
@@ -106,7 +106,7 @@ A4.0 conformance harness (delivered by PR #150):
   `generic_subprocess_runtime`; it does not perform runtime authority cutover. Issues #135/#136
   remain open until the later conformance, chaos, parity, and cutover slices are complete.
 
-A4.1a direct deterministic admission (this branch, CI-only):
+A4.1a direct deterministic admission (CI-only):
 
 - The `managed_runtime_direct_cutover` gate is an explicit opt-in dependency of
   `managed_runtime_worker` and is absent from every default profile. Bootstrap rejects it unless
@@ -118,10 +118,27 @@ A4.1a direct deterministic admission (this branch, CI-only):
 - Migration 0047 and the ORM check constraint reject an unbound managed admission while keeping
   legacy rows and explicit shadow admission compatible. Real PostgreSQL coverage verifies schema
   and repository round-trip behavior.
-- This slice does not connect the Worker to managed authority, change
-  `ManagedRuntimeExecutionService`, perform finalization/outcome reconciliation, enable the gate
-  on the test server, or claim production cutover. A4.1b (execution/authority) and A4.2
-  (reviewed/coordinated) remain open; issues #135/#136 remain open.
+- This slice does not enable the gate on the test server or claim production cutover.
+
+A4.1b.1 managed DIRECT Worker authority (CI/test-only):
+
+- The Worker selects only the immutable `TaskRun.runtime_authority`. Managed Runs never call the
+  legacy runner and continue on the managed path when the new-Run admission gate is later disabled.
+- Runtime prepare, fenced ownership, and the explicit `DISPATCHING` boundary use short
+  transactions; adapter/provider work remains outside a database transaction. Provider terminal
+  evidence and Task/Run/Attempt finalization, conservative budget settlement, quota release,
+  Inbox dedupe, and required Outbox events commit in one PostgreSQL UoW.
+- `OUTCOME_UNKNOWN`/`LOST`, provider response loss, and an expired owner after the dispatch
+  boundary park the execution in explicit Runtime/Task/Run/Attempt reconciliation states without
+  redispatch. Stable fenced uncertainty evidence makes message replay idempotent.
+- The slice accepts only mapping output with empty usage because pricing lineage is not yet part of
+  the managed observation path. Other successful provider results fail closed at the control-plane
+  result boundary. Real PostgreSQL tests cover atomic success/replay, injected rollback after
+  observation write, expired-owner parking, and stale-fence rollback.
+- No reconciliation command or ordinary transition leaves `RECONCILIATION_REQUIRED`; the
+  evidence-driven privileged convergence path is A4.1b.2. The gate remains absent from defaults
+  and disabled on the server. A4.2 reviewed/coordinated and A4.3 production durability/rollout
+  remain open; issues #135/#136 remain open.
 
 ## Current runnable baseline
 

docs/operations/runtime-direct-cutover-rollback.md

@@ -1,14 +1,14 @@
 # Deterministic direct-runtime admission rollback
 
-Last updated: 2026-08-20
+Last updated: 2026-08-21
 
-This runbook describes the A4.1a test-only admission gate. It is not a production runtime
-cutover procedure.
+This runbook describes the A4.1a/A4.1b.1 CI/test-only admission and managed DIRECT Worker path.
+It is not a production runtime cutover procedure.
 
 ## Scope
 
 `managed_runtime_direct_cutover` is disabled in all default profiles and must remain disabled on
-the test server until the later execution slice is explicitly qualified. When enabled, bootstrap
+the test server. When enabled, bootstrap
 accepts it only in `test`/`testing` with the deterministic model provider and requires
 `managed_runtime_worker`.
 
@@ -22,12 +22,16 @@ immutable persisted snapshot.
    `managed_runtime_direct_cutover=false`.
 2. Restart the API with the normal profile/configuration. The body-less task-run API remains
    available and new Runs use legacy authority.
-3. Do not rewrite existing `runtime_authority=managed` rows or change their Runtime Version. This
-   A4.1a slice has no Worker cutover, so those rows must not be claimed as evidence of a completed
-   managed execution path.
-4. Inspect the persisted Run and outbox records before any later execution slice is enabled.
-   Reconciliation or data repair is a separate, explicitly approved operation; never silently
-   retry by creating a new ordinary Run.
+3. Do not rewrite existing `runtime_authority=managed` rows or change their Runtime Version.
+   Disabling the admission gate affects new Runs only. Keep `managed_runtime_worker` and the
+   pinned built-in v2 adapter available until every existing managed Run is terminal or explicitly
+   parked for reconciliation; an existing managed Run must never fall back to legacy execution.
+4. Inspect RuntimeExecution phase, current owner/fence, latest Attempt, Inbox, and reconciliation
+   Outbox evidence. `DISPATCHING` or later with an expired owner and no reattach proof must park as
+   `RECONCILIATION_REQUIRED`; it must not be redispatched or replaced by an ordinary Run.
+5. A4.1b.1 deliberately has no exit from `RECONCILIATION_REQUIRED`. Escalate and preserve the
+   evidence until the privileged A4.1b.2 reconcile command is available. Manual status edits,
+   direct database repair, and blind provider retry are prohibited.
 
 Migration 0047 is expand-only and keeps legacy rows valid. A schema rollback is performed through
 the repository's tested Alembic downgrade window, not by manually dropping the managed admission

docs/roadmap.md

@@ -169,12 +169,15 @@ Exit signal：用户可从模板创建公司、绑定真实 Agent，在不伪造
 - [x] 非 LangGraph subprocess Agent 通过同一 conformance（A4.0，PR #150 已交付）
 - [x] A4.1a CI-only deterministic DIRECT 新 Run admission（gate、域模型、0047 持久化约束、
   内置 LangGraph v2 校验）
+- [x] A4.1b.1 CI/test-only managed DIRECT Worker authority（fenced dispatch、原子终结、
+  reconciliation-required 停车；不含 reconcile command）
 - [ ] MCP write 和 fake external action 通过统一 Intent/Permit/Receipt/Reconciliation
 - [ ] Chaos smoke 证明核心 crash windows 收敛且无重复不可逆副作用
 
 Exit signal：同一部署管理 LangGraph 与非 LangGraph Agent；两者使用同一 Task/Run/Attempt、
 身份、治理、Artifact 和恢复语义，并由机器可读故障报告证明关键不变量。
 
-当前 A4.0 conformance harness 已在 PR #150 完成，A4.1a 已交付 gate/domain/persistence/admission
-边界；两者均不满足完整 A4，也不关闭 #135/#136。A4.1b 仍需 Worker authority execution、
-finalization/outcome reconciliation，完整 A4 还需 chaos、parity 和 reviewed/coordinated cutover。
+当前 A4.0 conformance harness 已在 PR #150 完成，A4.1a admission 与 A4.1b.1 managed DIRECT
+Worker authority/atomic parking 已交付。A4.1b.2 仍需受权限控制、证据驱动的 reconcile command；
+完整 A4 还需 chaos、parity、reviewed/coordinated cutover 和生产 durable runtime。#135/#136
+继续保持开放。

src/agentmesh/application/managed_runtime_execution.py

@@ -10,10 +10,13 @@
 from __future__ import annotations
 
 from datetime import datetime, timezone
-from uuid import UUID
+from uuid import NAMESPACE_URL, UUID, uuid5
 
 from agentmesh.application.ports import (
+    ManagedRuntimeAuthoritativeResult,
+    ManagedRuntimeControlPlaneFailure,
     ManagedRuntimeExecutionPort,
+    ManagedRuntimePreDispatchFailure,
     RuntimeAssignmentBuilder,
     WorkflowWorkItem,
 )
@@ -22,7 +25,16 @@
 from agentmesh.domain.errors import InvalidTaskTransition
 from agentmesh.domain.runtime_execution import RuntimeExecutionPhase
 from agentmesh.domain.tasks import AttemptStatus, Task, TaskAttempt, TaskRun
-from agentmesh.runtime_sdk import ManagedAgentRuntime, RuntimeObservation
+from agentmesh.runtime_sdk import (
+    ErrorCategory,
+    ManagedAgentRuntime,
+    RetryDisposition,
+    RuntimeAssignment,
+    RuntimeError,
+    RuntimeObservation,
+    RuntimePhase,
+    canonical_digest,
+)
 
 _PHASES = {phase.value: phase for phase in RuntimeExecutionPhase}
 
@@ -150,12 +162,159 @@ def execute_shadow(
             evidence_id=observation.observation_id,
         )
 
+    def execute_authoritative(
+        self,
+        task: Task,
+        run: TaskRun,
+        attempt: TaskAttempt,
+        *,
+        work_item: WorkflowWorkItem | None = None,
+    ) -> ManagedRuntimeAuthoritativeResult:
+        """Dispatch one managed-authoritative execution without committing its outcome."""
+        execution_identity = run.runtime_execution_id or run.runtime_execution_intent_id
+        if run.runtime_authority != "managed" or run.runtime_version_id is None:
+            raise ValueError("Managed authority requires a pinned Runtime Version")
+        if execution_identity is None:
+            raise ValueError("Managed authority requires a Runtime execution intent")
+        now = datetime.now(timezone.utc)
+        if attempt.status is not AttemptStatus.RUNNING or _utc(attempt.lease_expires_at) <= now:
+            raise InvalidTaskTransition("Managed Runtime Attempt lease is not active")
+        try:
+            assignment = self._assignment_builder.assignment_for(
+                task, run, attempt, work_item=work_item
+            )
+            report = self._adapter.validate(assignment)
+            if not report.valid:
+                raise ValueError("Managed Runtime assignment validation failed")
+            binder = getattr(self._adapter, "bind_context", None)
+            if binder is None:
+                raise ValueError("Managed Runtime adapter has no assignment backend")
+            binder(assignment, task, run, attempt, work_item)
+        except Exception as exc:
+            raise ManagedRuntimePreDispatchFailure(
+                "Managed Runtime assignment preparation failed"
+            ) from exc
+        expected_key = f"runtime-dispatch:{task.tenant_id}:{execution_identity}"
+        try:
+            execution = self._registry.prepare_execution(
+                run_id=run.id,
+                assignment_id=_uuid(assignment.assignment_id),
+                assignment_digest=assignment.assignment_digest or "",
+                dispatch_key=expected_key,
+                execution_id=execution_identity,
+            )
+        except Exception as exc:
+            raise ManagedRuntimeControlPlaneFailure(
+                "Managed Runtime execution preparation did not commit"
+            ) from exc
+        if execution.phase is not RuntimeExecutionPhase.PREPARED:
+            return self._unknown_result(
+                execution.id,
+                assignment,
+                "runtime.reattach_unavailable",
+                observed_at=execution.updated_at,
+                dispatch_crossed=True,
+            )
+        try:
+            execution = self._registry.claim_execution_owner(
+                execution_id=execution.id,
+                attempt_id=attempt.id,
+                fencing_token=attempt.fencing_token,
+                expected_owner_attempt_id=execution.current_owner_attempt_id,
+                expected_fencing_token=execution.current_fencing_token,
+                expected_version=execution.version,
+                claim_reason=(
+                    "replacement"
+                    if execution.current_owner_attempt_id is not None
+                    else "initial"
+                ),
+                now=now,
+            )
+            execution = self._registry.mark_execution_dispatching(
+                execution_id=execution.id,
+                attempt_id=attempt.id,
+                fencing_token=attempt.fencing_token,
+            )
+        except Exception as exc:
+            raise ManagedRuntimeControlPlaneFailure(
+                "Managed Runtime dispatch boundary did not commit"
+            ) from exc
+        try:
+            receipt = self._adapter.dispatch(assignment, dispatch_key=expected_key)
+            observation = receipt.observation
+            if observation is None:
+                observation = self._adapter.inspect(receipt.handle)
+            self._validate_identity(execution.id, assignment, observation)
+        except Exception:
+            return self._unknown_result(
+                execution.id,
+                assignment,
+                "runtime.provider_outcome_unknown",
+                observed_at=execution.updated_at,
+                dispatch_crossed=True,
+            )
+        return ManagedRuntimeAuthoritativeResult(
+            execution_id=execution.id,
+            assignment_id=_uuid(assignment.assignment_id),
+            assignment_digest=assignment.assignment_digest or "",
+            observation=observation,
+            dispatch_crossed=True,
+        )
+
+    @staticmethod
+    def _validate_identity(
+        execution_id: UUID, assignment: RuntimeAssignment, observation: object
+    ) -> None:
+        if type(observation) is not RuntimeObservation:
+            raise ValueError("Runtime observation type is inconsistent")
+        assignment_id = assignment.assignment_id
+        assignment_digest = assignment.assignment_digest
+        if (
+            observation.runtime_execution_id != str(execution_id)
+            or observation.assignment_id != assignment_id
+            or observation.assignment_digest != assignment_digest
+            or not observation.phase.terminal
+        ):
+            raise ValueError("Runtime observation identity is inconsistent")
+
+    @staticmethod
+    def _unknown_result(
+        execution_id: UUID,
+        assignment: RuntimeAssignment,
+        code: str,
+        *,
+        observed_at: datetime,
+        dispatch_crossed: bool,
+    ) -> ManagedRuntimeAuthoritativeResult:
+        assignment_id = assignment.assignment_id
+        assignment_digest = assignment.assignment_digest
+        observation = RuntimeObservation(
+            observation_id=str(uuid5(NAMESPACE_URL, f"{execution_id}:{code}")),
+            runtime_execution_id=str(execution_id),
+            assignment_id=assignment_id,
+            assignment_digest=assignment_digest,
+            phase=RuntimePhase.OUTCOME_UNKNOWN,
+            observed_at=_utc(observed_at),
+            provider_event_id=code,
+            error=RuntimeError(
+                code=code,
+                category=ErrorCategory.UNKNOWN,
+                message="Runtime provider outcome requires reconciliation",
+                retry_disposition=RetryDisposition.RECONCILE,
+            ),
+        )
+        return ManagedRuntimeAuthoritativeResult(
+            execution_id=execution_id,
+            assignment_id=_uuid(assignment_id),
+            assignment_digest=assignment_digest,
+            observation=observation,
+            dispatch_crossed=dispatch_crossed,
+        )
+
 
 def _uuid(value: str):
     return UUID(value)
 
 
 def _observation_digest(observation: RuntimeObservation) -> str:
-    from agentmesh.runtime_sdk import canonical_digest
-
     return canonical_digest(observation.to_dict())

src/agentmesh/application/ports.py

@@ -101,7 +101,7 @@
     ToolExecutionAuthorization,
     ToolInvocation,
 )
-from agentmesh.runtime_sdk import RuntimeAssignment
+from agentmesh.runtime_sdk import RuntimeAssignment, RuntimeObservation
 
 
 class TaskRepository(Protocol):
@@ -1169,6 +1169,23 @@ def assignment_for(
     ) -> RuntimeAssignment: ...
 
 
+@dataclass(frozen=True)
+class ManagedRuntimeAuthoritativeResult:
+    execution_id: UUID
+    assignment_id: UUID
+    assignment_digest: str
+    observation: RuntimeObservation
+    dispatch_crossed: bool
+
+
+class ManagedRuntimePreDispatchFailure(RuntimeError):
+    """Assignment was rejected before persistent execution/provider effects."""
+
+
+class ManagedRuntimeControlPlaneFailure(RuntimeError):
+    """A recoverable prepare/claim/dispatch-boundary transaction failed."""
+
+
 class ManagedRuntimeExecutionPort(Protocol):
     """A transactional coordinator around the framework-neutral runtime port."""
 
@@ -1181,6 +1198,15 @@ def execute_shadow(
         work_item: WorkflowWorkItem | None = None,
     ) -> Any: ...
 
+    def execute_authoritative(
+        self,
+        task: Task,
+        run: TaskRun,
+        attempt: TaskAttempt,
+        *,
+        work_item: WorkflowWorkItem | None = None,
+    ) -> ManagedRuntimeAuthoritativeResult: ...
+
 
 @dataclass(frozen=True)
 class AgentExecutionContext: