docs(codex): publish V2R1 G1 final verification retry

.agent/results/codex/RESULT-V2R1-202607132000-CODEX.md

@@ -0,0 +1,38 @@
+status: PASS
+integration_path: tests/integration/test_runtime_ingress_flow.py
+tests: pytest tests/v2r1 = 29 passed in 0.98s; pytest tests/integration = 3 passed in 0.18s
+runtime_flow: PASS - signed public ingress -> IngressAuthenticator -> RuntimeIdentityResolver -> RuntimeWiring.ingest -> ChannelAdapter -> MessageLedger.append -> SessionPool
+commit: a6878d30b8690e55f1bf9819da5aac189dabf89e
+gate1: BLOCKED
+risk: high
+
+# Result
+
+## Changed Files
+
+- `tests/integration/test_runtime_ingress_flow.py`
+- `.agent/results/codex/RESULT-V2R1-202607132000-CODEX.md`
+
+## RED Evidence
+
+The baseline integration entrypoint from `f11513e9d7326caa827e59f4f10f07681633dfc0` was inadequate: its successful flow replaced `MessageLedger.remember()` and its alleged duplicate conflict changed the message ID.  The quality checks failed before this update because neither test proved the required runtime contract.
+
+## Verification
+
+- The success path sends a real signed public ingress request and uses a `sys.setprofile` observer against the real `MessageLedger.remember` code object.  It preserves the production object and all behavior, then asserts zero legacy calls.
+- The duplicate path sends two authenticated requests under the same principal, tenant, cell, channel, account, peer, direction, session, and `message_id`; only `event_id` and body differ.  The second response must be `CONFLICT`, with exactly one ledger record and one session lease remaining.
+
+## Commands
+
+```text
+PYTHONDONTWRITEBYTECODE=1 PYTHONPATH=active-runtime/app /usr/bin/python3 -m pytest tests/v2r1 -q
+PYTHONDONTWRITEBYTECODE=1 PYTHONPATH=active-runtime/app /usr/bin/python3 -m pytest tests/integration -q
+```
+
+## Gate 1
+
+`BLOCKED`. This task makes the runtime integration entrypoint meaningful, but does not close the independent Gate 1 matrix, authenticated production OpenClaw registration, secret scanning, second clean-room review, or staging rollback rehearsal.
+
+## Recommendation
+
+Keep this isolated evidence commit for independent review. Do not merge, tag, or deploy until all Gate 1 requirements have independent evidence.

.agent/results/codex/RESULT-V2R1-G1-CODEX-CLEANUP-20260713.md

@@ -0,0 +1,95 @@
+status: PASS
+commit: 7b82ae11f685b77628308ba46b9cfdf5bcdf78d3
+tests:
+  v2r1: "29 passed, exit code 0"
+  integration: "3 passed, exit code 0"
+cleanup_scan:
+  status: PASS
+  findings: "No remaining conflict markers, unexplained worktree changes, unapproved temporary/cache/build artifacts, or scanner-detected secrets. Runtime SQLite was removed and its sidecars are ignored with exact root-anchored rules; tracked dist/ZIP artifacts are documented exceptions."
+risk: MEDIUM
+recommendation: "Push the controlled work branch, then request independent OpenClaw Gate Review; do not merge or release from this result alone."
+
+# V2R1 Gate 1 Codex Cleanup Result
+
+## Scope and Git Baseline
+
+- Task: `V2R1-G1-CODEX-CLEANUP-20260713`
+- Branch: `work/v2r1-g1-cleanup-remediation-20260713`
+- Cleanup implementation commit: `7b82ae11f685b77628308ba46b9cfdf5bcdf78d3`
+- Integration commit: `a6878d30b8690e55f1bf9819da5aac189dabf89e`
+- Locked integration result baseline: `d6ba7e8fa859407ec55acf49b5468bba5eac96da`
+- `git merge-base --is-ancestor a6878d30b8690e55f1bf9819da5aac189dabf89e HEAD`: exit code `0`
+- `git merge-base --is-ancestor d6ba7e8fa859407ec55acf49b5468bba5eac96da HEAD`: exit code `0`
+- Tree before this result-only commit: `94146e05f700921b12ef023eb4d8be771bae4047`
+
+The locked integration artifacts were not modified by cleanup:
+
+- `tests/integration/test_runtime_ingress_flow.py`: blob `6b6732a4c73bbb7fb5273a05b897adce9da39a9a`
+- `.agent/results/codex/RESULT-V2R1-202607132000-CODEX.md`: blob `e2d6ce3b037339385831f51cb0db42c86b2ae827`
+
+## Cleanup Changes Verified
+
+Commit `7b82ae11f685b77628308ba46b9cfdf5bcdf78d3`:
+
+- removed the tracked runtime database `data/webops.sqlite3`;
+- added exact root-anchored ignore rules for `/data/webops.sqlite3`, `/data/webops.sqlite3-wal`, `/data/webops.sqlite3-shm`, and `/data/webops.sqlite3-journal`;
+- resolved tracked merge-conflict residue in `agents/registry.yaml` and `docs/architecture/AIOS_AGENT_AUDIT_PLANE.md`;
+- left the locked integration test and prior Codex result blobs unchanged.
+
+`git grep -n -E '^(<<<<<<<|=======|>>>>>>>)' -- . ':!source-baseline/*.zip'` returned no matches (exit code `1`, the expected no-match result).
+
+The repository scan found no unapproved `__pycache__`, `*.pyc`, `*.pyo`, `.pytest_cache`, `.mypy_cache`, `.ruff_cache`, `*.tmp`, `*.swp`, backup-suffix, or `.DS_Store` artifacts after verification cleanup. The following tracked artifacts are expected baseline exceptions, not unexplained generated files:
+
+- `openclaw/plugins/aios-tenant-boundary/dist/` (2 tracked plugin distribution files);
+- `source-baseline/current_actual_deployment_source.zip`;
+- `source-baseline/previous_current_actual_deployment_source_from_webops_delta.zip`.
+
+Before creating this result file, `git status --porcelain=v1 --untracked-files=all` was empty with exit code `0`.
+
+## Secret Scan
+
+Scanner: Gitleaks `8.30.1`, official Linux x64 release asset.
+
+- Asset: `/tmp/gitleaks_8.30.1_linux_x64.tar.gz.codex`
+- SHA-256: `551f6fc83ea457d62a0d98237cbad105af8d557003051f41f3e7ca7b3f2470eb`
+- `/tmp/gitleaks-codex-8.30.1/gitleaks git --no-banner --no-color --redact --report-format json --report-path /tmp/v2r1-cleanup-audit-git.json --log-opts=7b82ae11f685b77628308ba46b9cfdf5bcdf78d3 /data/aios-gitops/worktrees/v2r1-g1-cleanup-remediation-20260713`: exit code `0`; Gitleaks reported `312 commits scanned`, approximately `1,384,427 bytes (1.38 MB)` scanned, and `no leaks found` (`0` findings in the JSON report).
+- `git rev-list --count 7b82ae11f685b77628308ba46b9cfdf5bcdf78d3`: exit code `0`, `327` reachable commits. This Git reachability count is recorded separately and is not the Gitleaks scanned-commit count.
+- `/tmp/gitleaks-codex-8.30.1/gitleaks dir --no-banner --no-color --redact --report-format json --report-path /tmp/v2r1-cleanup-audit-tree.json /data/aios-gitops/worktrees/v2r1-g1-cleanup-remediation-20260713`: exit code `0`; Gitleaks reported approximately `1,214,737 bytes (1.21 MB)` scanned and `no leaks found` (`0` findings in the JSON report).
+
+No secret content is reproduced in this report.
+
+## Tests
+
+Command:
+
+```bash
+PYTHONDONTWRITEBYTECODE=1 PYTHONPATH=active-runtime/app /usr/bin/python3 -m pytest tests/v2r1
+```
+
+Output summary:
+
+```text
+collected 29 items
+29 passed in 0.97s
+exit code: 0
+```
+
+Command:
+
+```bash
+PYTHONDONTWRITEBYTECODE=1 PYTHONPATH=active-runtime/app /usr/bin/python3 -m pytest tests/integration
+```
+
+Output summary:
+
+```text
+collected 3 items
+3 passed in 0.18s
+exit code: 0
+```
+
+Environment: Linux, Python `3.12.3`, pytest `9.1.1`, pluggy `1.6.0`.
+
+## Disposition
+
+The cleanup and requested test verification pass at the stated implementation commit. Risk remains `MEDIUM` until the controlled branch is pushed and independently fetched and reviewed by OpenClaw. This result does not authorize merge, tag creation, baseline update, staging, or production deployment.

.agent/results/codex/RESULT-V2R1-G1-CODEX-FINAL-VERIFY-RETRY-20260713.md

@@ -0,0 +1,108 @@
+---
+status: PASS
+task_id: TASK-CODEX-V2R1-G1-FINAL-VERIFY-RETRY-20260715T010807UTC8
+verification_branch: review/codex/v2r1-g1-final-verify-retry-20260715T010807UTC8
+source_commit: 502ef30642597fb0b70b90ae2e8a0b6af217d80e
+candidate_commit: bfd13bbed0d17a687283f0eec59c1d87a3e6a927
+candidate_branch: work/codex/v2r1-g1-recovery-candidate-20260715T010807UTC8-r2
+qualifying_candidates:
+  - bfd13bbed0d17a687283f0eec59c1d87a3e6a927
+rejected:
+  - commit: 7a2f16ee72e944737a9f6995e511ef176413ae6b
+    reason: permanent_rejection
+  - commit: b1d3439ee9e66b5eed1c94595a22bc630d893b6e
+    reason: permanent_rejection
+  - commit: 38320044b54c172c28877a81f291478267dd7b9e
+    reason: permanent_rejection
+  - commit: 7fd63b55f27b23658ea26e2a39d949c7b1e91dda
+    reason: permanent_rejection
+  - commit: 71239b4d187e683e05fd22fc9d5df8ae324e3390
+    reason: permanent_rejection
+  - commit: 78c22cbfb95cb252b671c5ffe3b34bf4944aee2d
+    reason: invalid_placeholder_count_1
+placeholder_scan:
+  status: PASS
+  token_count: 0
+  angle_count: 0
+  total_count: 0
+placeholder_count: 0
+mandatory_fields: true
+mandatory_field_names:
+  - status
+  - source_commit
+  - candidate_commit
+  - qualifying_candidates
+  - placeholder_count
+  - mandatory_fields
+  - tests
+  - gitleaks
+  - hmac
+  - cleanup_scan
+  - rollback
+  - deployment
+  - workspace_clean
+  - risk
+  - recommendation
+tests:
+  status: PASS
+  v2r1:
+    command: "PYTHONDONTWRITEBYTECODE=1 PYTEST_ADDOPTS='-p no:cacheprovider' PYTHONPATH=active-runtime/app /usr/bin/python3 -m pytest tests/v2r1 -q"
+    exit_code: 0
+    passed: 29
+  integration:
+    command: "PYTHONDONTWRITEBYTECODE=1 PYTEST_ADDOPTS='-p no:cacheprovider' PYTHONPATH=active-runtime/app /usr/bin/python3 -m pytest tests/integration -q"
+    exit_code: 0
+    passed: 3
+source_gates:
+  conflict_count: 0
+  tracked_sqlite_count: 0
+  ordered_webops_ignore_rule_count: 4
+gitleaks:
+  status: PASS
+  version: "8.30.1"
+  git:
+    exit_code: 0
+    findings: 0
+  dir:
+    exit_code: 0
+    findings: 0
+hmac:
+  status: PASS
+  internal_key_length: 39
+  ingress_key_length: 39
+runtime_secret_acl:
+  status: PASS
+  service_identity: aiman:gitops
+  directory:
+    path: /data/aios-gitops/runtime/secrets
+    owner: root:gitops
+    mode: 750
+  file:
+    path: /data/aios-gitops/runtime/secrets/gitops.tokens
+    owner: root:gitops
+    mode: 640
+  service_access:
+    read: PASS
+    write: REJECTED
+    replace: REJECTED
+secret_scan: PASS
+cleanup_scan:
+  status: PASS
+  temporary_reports_removed: true
+  unexpected_workspace_artifacts: 0
+rollback:
+  status: PASS
+  required: false
+  mechanism: Remove the review ref before promotion if the owner gate rejects it.
+deployment: false
+merge: false
+tag: false
+baseline: false
+workspace_clean: true
+risk: MEDIUM
+risk_details:
+  interactive_aiman_sudo_residual: true
+  interactive_aiman_docker_residual: true
+  unrestricted_passwordless_sudo: true
+recommendation: Maintain the owner gate; merge, tag, baseline change, and deployment are not authorized.
+---

.agent/results/openclaw/RESULT-V2R1-G1-OPENCLAW-RECOVERY-CANDIDATE-20260715T010807UTC8.md

@@ -0,0 +1,124 @@
+---
+status: PASS
+commit: 502ef30642597fb0b70b90ae2e8a0b6af217d80e
+task_id: TASK-CODEX-V2R1-G1-FINAL-VERIFY-RETRY-20260715T010807UTC8
+executor: CODEX_CANDIDATE_BUILDER
+recovery_iteration: r2
+rejected_candidate: 78c22cbfb95cb252b671c5ffe3b34bf4944aee2d
+source_commit: 502ef30642597fb0b70b90ae2e8a0b6af217d80e
+source_branch: work/v2r1-integration-quality-2000
+candidate_ref: work/codex/v2r1-g1-recovery-candidate-20260715T010807UTC8-r2
+tests:
+  status: PASS
+  v2r1:
+    command: "PYTHONDONTWRITEBYTECODE=1 PYTEST_ADDOPTS='-p no:cacheprovider' PYTHONPATH=active-runtime/app /usr/bin/python3 -m pytest tests/v2r1 -q"
+    exit_code: 0
+    passed: 29
+    summary: 29 passed in 0.95s
+  integration:
+    command: "PYTHONDONTWRITEBYTECODE=1 PYTEST_ADDOPTS='-p no:cacheprovider' PYTHONPATH=active-runtime/app /usr/bin/python3 -m pytest tests/integration -q"
+    exit_code: 0
+    passed: 3
+    summary: 3 passed in 0.19s
+commands:
+  - id: v2r1_tests
+    exit_code: 0
+  - id: integration_tests
+    exit_code: 0
+  - id: source_object_conflict_marker_scan
+    exit_code: 1
+    accepted_exit_code: 1
+  - id: tracked_sqlite_object_scan
+    command: "git ls-tree -r --name-only 502ef30642597fb0b70b90ae2e8a0b6af217d80e -- data/webops.sqlite3 data/webops.sqlite3-wal data/webops.sqlite3-shm data/webops.sqlite3-journal"
+    exit_code: 0
+  - id: ordered_webops_ignore_rule_scan
+    exit_code: 0
+  - id: gitleaks_version
+    binary: /tmp/gitleaks-codex-8.30.1/gitleaks
+    exit_code: 0
+  - id: gitleaks_git_source_object_scan
+    exit_code: 0
+  - id: gitleaks_directory_scan
+    exit_code: 0
+  - id: python_ast_key_length_scan
+    exit_code: 0
+conflict_scan:
+  status: PASS
+  command_name: source_object_conflict_marker_scan
+  source_object: 502ef30642597fb0b70b90ae2e8a0b6af217d80e
+  exit_code: 1
+  stdout_lines: 0
+  conflict_count: 0
+tracked_sqlite:
+  status: PASS
+  exit_code: 0
+  count: 0
+  checked_paths:
+    - data/webops.sqlite3
+    - data/webops.sqlite3-wal
+    - data/webops.sqlite3-shm
+    - data/webops.sqlite3-journal
+webops_ignore_rules:
+  status: PASS
+  command_name: ordered_webops_ignore_rule_scan
+  count: 4
+  ordered:
+    - /data/webops.sqlite3
+    - /data/webops.sqlite3-wal
+    - /data/webops.sqlite3-shm
+    - /data/webops.sqlite3-journal
+gitleaks:
+  status: PASS
+  version: "8.30.1"
+  git:
+    command_name: gitleaks_git_source_object_scan
+    source_object: 502ef30642597fb0b70b90ae2e8a0b6af217d80e
+    exit_code: 0
+    findings: 0
+    report_json: []
+  dir:
+    command_name: gitleaks_directory_scan
+    exit_code: 0
+    findings: 0
+    report_json: []
+hmac:
+  status: PASS
+  method: AST assignment inspection
+  target: tests/v2r1/test_runtime_wiring.py
+  command_name: python_ast_key_length_scan
+  exit_code: 0
+  internal_key_length: 39
+  ingress_key_length: 39
+placeholder_scan:
+  status: PASS
+  token_count: 0
+  angle_count: 0
+  total_count: 0
+placeholder_count: 0
+mandatory_fields:
+  status: PASS
+  present:
+    - status
+    - commit
+    - tests
+    - cleanup_scan
+    - risk
+    - recommendation
+cleanup_scan:
+  status: PASS
+  details:
+    temporary_reports_outside_worktree: true
+    temporary_reports_removed: true
+    unexpected_workspace_artifacts: 0
+secret_scan: PASS
+rollback:
+  status: PASS
+  required: false
+deployment: false
+merge: false
+tag: false
+baseline: false
+workspace_clean_at_test_completion: true
+risk: LOW
+recommendation: Advance this r2 candidate commit to independent verification only; do not deploy, merge, tag, or alter baselines.
+---

.gitignore

@@ -11,3 +11,7 @@ runtime/tmp/
 runtime/cache/
 logs/
 config/test_write_perms
+/data/webops.sqlite3
+/data/webops.sqlite3-wal
+/data/webops.sqlite3-shm
+/data/webops.sqlite3-journal

active-runtime/app/aios_v2r1/channel_adapter.py

@@ -0,0 +1,64 @@
+"""The only V2R1 ingress adapter: verified context to routed ledger record."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime
+from threading import RLock
+from typing import Callable, Mapping, Any
+
+from .message_ledger import LedgerRecord, MessageLedger
+from .router import Route, TenantRouter
+from .session_pool import SessionLease, SessionPool
+from .tenant_context import CanonicalMessage, TenantContext, utc_now
+
+
+@dataclass(frozen=True)
+class IngressResult:
+    record: LedgerRecord
+    route: Route
+    session: SessionLease
+
+
+class ChannelAdapter:
+    """Fail-closed runtime ingress composition.
+
+    The adapter intentionally accepts no tenant or Cell authority outside the
+    signed context and TenantRouter.  A process lock keeps rollback handling
+    deterministic for a single runtime while SQLite protects cross-process
+    ledger writes.
+    """
+
+    def __init__(
+        self,
+        *,
+        router: TenantRouter,
+        ledger: MessageLedger,
+        session_pool: SessionPool,
+        signing_key: bytes,
+        clock: Callable[[], datetime] = utc_now,
+    ) -> None:
+        self._router = router
+        self._ledger = ledger
+        self._session_pool = session_pool
+        self._signing_key = signing_key
+        self._clock = clock
+        self._lock = RLock()
+
+    def ingest(self, *, token: str, message: Mapping[str, Any]) -> IngressResult:
+        with self._lock:
+            context = TenantContext.verify(token, self._signing_key, now=self._clock())
+            canonical = CanonicalMessage.from_mapping(message)
+            canonical.assert_context(context)
+            route = self._router.assert_context_route(context)
+            self._router.assert_message_route(canonical)
+            self._session_pool.assert_available(context, route)
+
+            existed = self._ledger.matches(canonical)
+            record = self._ledger.append(canonical)
+            try:
+                lease = self._session_pool.acquire(context, route)
+            except Exception:
+                if not existed:
+                    self._ledger.remove(record)
+                raise
+            return IngressResult(record=record, route=route, session=lease)

active-runtime/app/aios_v2r1/hermes_fabric.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 
 from dataclasses import dataclass
 from datetime import datetime, timezone
+from contextlib import closing
 from pathlib import Path
 import sqlite3
 from typing import Any
@@ -42,7 +43,7 @@ class HermesFabric:
 
     def __init__(self, database: str | Path) -> None:
         self.database = str(database)
-        with sqlite3.connect(self.database) as db:
+        with closing(sqlite3.connect(self.database)) as connection, connection as db:
             db.execute("PRAGMA journal_mode=WAL")
             db.execute("PRAGMA synchronous=FULL")
             db.execute(
@@ -87,7 +88,7 @@ class HermesFabric:
         now = isoformat()
         conflict: ConflictError | None = None
         try:
-            with sqlite3.connect(self.database) as db:
+            with closing(sqlite3.connect(self.database)) as connection, connection as db:
                 db.execute("PRAGMA journal_mode=WAL")
                 db.execute(
                     "INSERT INTO fabric_jobs VALUES(?,?,?,?,?,?,?,?,?)",
@@ -115,7 +116,7 @@ class HermesFabric:
         require_ident("tenant_id", tenant_id, tenant=True)
         require_ident("cell_id", cell_id)
         require_ident("profile_id", profile_id)
-        with sqlite3.connect(self.database) as db:
+        with closing(sqlite3.connect(self.database)) as connection, connection as db:
             db.execute("BEGIN IMMEDIATE")
             row = db.execute(
                 "SELECT * FROM fabric_jobs WHERE tenant_id=? AND cell_id=? "
@@ -133,7 +134,7 @@ class HermesFabric:
         return FabricJob(*row[:7])
 
     def health(self) -> dict[str, Any]:
-        with sqlite3.connect(self.database) as db:
+        with closing(sqlite3.connect(self.database)) as connection, connection as db:
             count = db.execute("SELECT count(*) FROM fabric_jobs").fetchone()[0]
         return {
             "component": "hermes_fabric",

active-runtime/app/aios_v2r1/identity.py

@@ -0,0 +1,158 @@
+"""Fail-closed identity binding for authenticated external ingress.
+
+External payload claims are not authority.  An HMAC-authenticated envelope is
+the sole factory for :class:`AuthenticatedPrincipal`; the resolver then checks
+every claim against the authoritative ``TenantRouter`` before the runtime signs
+an internal ``TenantContext``.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+import hashlib
+import hmac
+import re
+from typing import Any, Mapping
+
+from .router import Route, TenantRouter
+from .tenant_context import ScopeError, ValidationError, canonical_json, require_ident
+
+
+_PROVENANCE = object()
+_SIGNATURE_RE = re.compile(r"^[0-9a-f]{64}$")
+_SCOPE_FIELDS = (
+    "principal_id", "tenant_id", "cell_id", "channel",
+    "channel_account_id", "session_id", "message",
+)
+
+
+class AuthenticationError(ValidationError):
+    """Authentication or identity resolution failed without disclosing scope."""
+
+
+@dataclass(frozen=True, init=False)
+class AuthenticatedPrincipal:
+    """A principal which only the verified envelope factory can create."""
+
+    principal_id: str
+
+    def __init__(self, principal_id: str, *, _provenance: object) -> None:
+        if _provenance is not _PROVENANCE:
+            raise TypeError("AuthenticatedPrincipal is created by authenticated ingress only")
+        object.__setattr__(self, "principal_id", require_ident("principal_id", principal_id))
+
+    @classmethod
+    def _verified(cls, principal_id: str) -> "AuthenticatedPrincipal":
+        return cls(principal_id, _provenance=_PROVENANCE)
+
+
+@dataclass(frozen=True)
+class IdentityClaims:
+    tenant_id: str
+    cell_id: str
+    channel: str
+    channel_account_id: str
+    session_id: str
+
+    def __post_init__(self) -> None:
+        require_ident("tenant_id", self.tenant_id, tenant=True)
+        for field in ("cell_id", "channel", "channel_account_id", "session_id"):
+            require_ident(field, getattr(self, field))
+
+
+@dataclass(frozen=True)
+class AuthenticatedIngress:
+    principal: AuthenticatedPrincipal
+    claims: IdentityClaims
+    message: Mapping[str, Any]
+
+
+@dataclass(frozen=True)
+class RuntimeIdentity:
+    tenant_id: str
+    cell_id: str
+    principal_id: str
+    channel: str
+    channel_account_id: str
+    session_id: str
+
+
+class IngressAuthenticator:
+    """Verify a canonical HMAC envelope before any identity lookup occurs."""
+
+    def __init__(self, hmac_key: bytes) -> None:
+        if not isinstance(hmac_key, bytes) or len(hmac_key) < 32:
+            raise ValidationError("ingress HMAC key must be at least 32 bytes")
+        self._hmac_key = hmac_key
+
+    @staticmethod
+    def _signed_payload(request: Mapping[str, Any]) -> dict[str, Any]:
+        if not isinstance(request, Mapping):
+            raise ValidationError("ingress request must be an object")
+        if set(request) != {*_SCOPE_FIELDS, "signature"}:
+            raise ValidationError("ingress request fields are invalid")
+        values = {field: request[field] for field in _SCOPE_FIELDS}
+        if not isinstance(values["message"], Mapping):
+            raise ValidationError("ingress message must be an object")
+        return values
+
+    def authenticate(self, request: Mapping[str, Any]) -> AuthenticatedIngress:
+        values = self._signed_payload(request)
+        signature = request["signature"]
+        if not isinstance(signature, str) or not _SIGNATURE_RE.fullmatch(signature):
+            raise AuthenticationError("invalid ingress signature")
+        expected = hmac.new(
+            self._hmac_key, canonical_json(values), hashlib.sha256
+        ).hexdigest()
+        if not hmac.compare_digest(expected, signature):
+            raise AuthenticationError("invalid ingress signature")
+        return AuthenticatedIngress(
+            principal=AuthenticatedPrincipal._verified(values["principal_id"]),
+            claims=IdentityClaims(
+                tenant_id=values["tenant_id"],
+                cell_id=values["cell_id"],
+                channel=values["channel"],
+                channel_account_id=values["channel_account_id"],
+                session_id=values["session_id"],
+            ),
+            message=dict(values["message"]),
+        )
+
+
+class RuntimeIdentityResolver:
+    """Resolve authenticated ingress identity through ``TenantRouter`` only."""
+
+    def __init__(self, router: TenantRouter) -> None:
+        self._router = router
+
+    def resolve(
+        self, principal: AuthenticatedPrincipal, claims: IdentityClaims
+    ) -> RuntimeIdentity:
+        if not isinstance(principal, AuthenticatedPrincipal):
+            raise AuthenticationError("untrusted principal type")
+        try:
+            route: Route = self._router.resolve(
+                channel=claims.channel,
+                channel_account_id=claims.channel_account_id,
+                principal_id=principal.principal_id,
+                session_id=claims.session_id,
+            )
+        except (ScopeError, ValidationError) as exc:
+            raise AuthenticationError("identity is not registered") from None
+        expected = (
+            claims.tenant_id, claims.cell_id, claims.channel,
+            claims.channel_account_id, principal.principal_id, claims.session_id,
+        )
+        actual = (
+            route.tenant_id, route.cell_id, route.channel,
+            route.channel_account_id, route.principal_id, route.session_id,
+        )
+        if actual != expected:
+            raise AuthenticationError("identity claims do not match authoritative route")
+        return RuntimeIdentity(
+            tenant_id=route.tenant_id,
+            cell_id=route.cell_id,
+            principal_id=route.principal_id,
+            channel=route.channel,
+            channel_account_id=route.channel_account_id,
+            session_id=route.session_id or "",
+        )

active-runtime/app/aios_v2r1/message_ledger.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 
 from dataclasses import dataclass
 from datetime import timedelta
+from contextlib import closing
 import os
 from pathlib import Path
 import sqlite3
@@ -63,7 +64,7 @@ class MessageLedger:
         return db
 
     def _initialize(self) -> None:
-        with self._connect() as db:
+        with closing(self._connect()) as connection, connection as db:
             db.executescript(
                 """
                 CREATE TABLE IF NOT EXISTS message_ledger (
@@ -92,7 +93,13 @@ class MessageLedger:
                 """
             )
 
-    def remember(self, message: CanonicalMessage) -> LedgerRecord:
+    def append(self, message: CanonicalMessage) -> LedgerRecord:
+        """Append one exact message through the runtime write path.
+
+        This is intentionally the owner of validation, idempotency, and the
+        immediate SQLite transaction.  Runtime callers must not depend on the
+        legacy ``remember`` compatibility name.
+        """
         message_id = normalize_message_id(message.message_id)
         now = utc_now()
         expires = isoformat(now + timedelta(days=self.ttl_days))
@@ -105,27 +112,72 @@ class MessageLedger:
             message.direction,
             message_id,
         )
-        with self._connect() as db:
-            db.execute("BEGIN IMMEDIATE")
-            existing = db.execute(
-                "SELECT event_id, body, occurred_at, expires_at FROM message_ledger "
-                "WHERE tenant_id=? AND cell_id=? AND channel=? AND channel_account_id=? "
-                "AND peer_id=? AND direction=? AND message_id=?",
-                key,
-            ).fetchone()
-            if existing:
-                if existing["event_id"] != message.event_id or existing["body"] != message.body:
-                    db.rollback()
-                    raise ConflictError("message ID collision with different content")
+        storage_error: ConflictError | None = None
+        try:
+            with closing(self._connect()) as connection, connection as db:
+                db.execute("BEGIN IMMEDIATE")
+                existing = db.execute(
+                    "SELECT event_id, body, occurred_at, expires_at FROM message_ledger "
+                    "WHERE tenant_id=? AND cell_id=? AND channel=? AND channel_account_id=? "
+                    "AND peer_id=? AND direction=? AND message_id=?",
+                    key,
+                ).fetchone()
+                if existing:
+                    if existing["body"] != message.body:
+                        db.rollback()
+                        raise ConflictError("message ID collision with different content")
+                    db.commit()
+                    return LedgerRecord(
+                        *key, existing["event_id"], existing["body"],
+                        existing["occurred_at"], existing["expires_at"],
+                    )
+                db.execute(
+                    "INSERT INTO message_ledger VALUES(?,?,?,?,?,?,?,?,?,?,?,?)",
+                    (*key, message.event_id, message.body, message.occurred_at, expires, isoformat(now)),
+                )
                 db.commit()
-                return LedgerRecord(*key, existing["event_id"], existing["body"], existing["occurred_at"], existing["expires_at"])
-            db.execute(
-                "INSERT INTO message_ledger VALUES(?,?,?,?,?,?,?,?,?,?,?,?)",
-                (*key, message.event_id, message.body, message.occurred_at, expires, isoformat(now)),
-            )
-            db.commit()
+        except sqlite3.Error:
+            storage_error = ConflictError("ledger storage unavailable")
+        if storage_error is not None:
+            # Raise outside the sqlite exception handler so callers cannot inspect
+            # a driver exception through __cause__ or __context__.
+            raise storage_error
         return LedgerRecord(*key, message.event_id, message.body, message.occurred_at, expires)
 
+    def remember(self, message: CanonicalMessage) -> LedgerRecord:
+        """Backward-compatible API wrapper for callers outside the runtime path."""
+        return self.append(message)
+
+    def matches(self, message: CanonicalMessage) -> bool:
+        """Whether this exact message is already durably represented."""
+        message_id = normalize_message_id(message.message_id)
+        with closing(self._connect()) as connection, connection as db:
+            row = db.execute(
+                "SELECT event_id, body FROM message_ledger WHERE tenant_id=? AND "
+                "cell_id=? AND channel=? AND channel_account_id=? AND peer_id=? "
+                "AND direction=? AND message_id=?",
+                (
+                    message.tenant_id, message.cell_id, message.channel,
+                    message.channel_account_id, message.peer_id, message.direction,
+                    message_id,
+                ),
+            ).fetchone()
+        return bool(row and row["event_id"] == message.event_id and row["body"] == message.body)
+
+    def remove(self, record: LedgerRecord) -> None:
+        """Compensate a newly appended record if a later ingress step fails."""
+        with closing(self._connect()) as connection, connection as db:
+            db.execute(
+                "DELETE FROM message_ledger WHERE tenant_id=? AND cell_id=? AND "
+                "channel=? AND channel_account_id=? AND peer_id=? AND direction=? "
+                "AND message_id=? AND event_id=?",
+                (
+                    record.tenant_id, record.cell_id, record.channel,
+                    record.channel_account_id, record.peer_id, record.direction,
+                    record.message_id, record.event_id,
+                ),
+            )
+
     def resolve_quote(self, message: CanonicalMessage) -> str:
         if message.quoted_body is not None:
             return message.quoted_body
@@ -135,7 +187,7 @@ class MessageLedger:
             message.reply_to_message_id,
             field_name="reply_to_message_id",
         )
-        with self._connect() as db:
+        with closing(self._connect()) as connection, connection as db:
             rows = db.execute(
                 "SELECT direction, body, event_id FROM message_ledger "
                 "WHERE tenant_id=? AND cell_id=? AND channel=? AND channel_account_id=? "
@@ -157,16 +209,16 @@ class MessageLedger:
         raise ScopeError("quoted_text_unavailable")
 
     def purge_expired(self) -> int:
-        with self._connect() as db:
+        with closing(self._connect()) as connection, connection as db:
             cursor = db.execute(
                 "DELETE FROM message_ledger WHERE expires_at<=?", (isoformat(),)
             )
             return cursor.rowcount
 
     def integrity_check(self) -> str:
-        with self._connect() as db:
+        with closing(self._connect()) as connection, connection as db:
             return str(db.execute("PRAGMA integrity_check").fetchone()[0])
 
     def count(self) -> int:
-        with self._connect() as db:
+        with closing(self._connect()) as connection, connection as db:
             return int(db.execute("SELECT count(*) FROM message_ledger").fetchone()[0])

active-runtime/app/aios_v2r1/router.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 from contextlib import contextmanager
+from contextlib import closing
 from dataclasses import dataclass
 from datetime import datetime, timedelta
 import sqlite3
@@ -69,7 +70,7 @@ class TenantRouter:
             connection.close()
 
     def _initialize(self) -> None:
-        with self._connect() as db:
+        with closing(self._connect()) as connection, connection as db:
             db.executescript(
                 """
                 CREATE TABLE IF NOT EXISTS tenants (
@@ -257,7 +258,7 @@ class TenantRouter:
     def resolve_channel(self, channel: str, channel_account_id: str) -> Route:
         channel = require_ident("channel", channel)
         channel_account_id = require_ident("channel_account_id", channel_account_id)
-        with self._connect() as db:
+        with closing(self._connect()) as connection, connection as db:
             row = db.execute(
                 "SELECT r.tenant_id, r.cell_id, t.status FROM channel_routes r "
                 "JOIN tenants t ON t.tenant_id=r.tenant_id "
@@ -280,7 +281,7 @@ class TenantRouter:
     ) -> Route:
         principal_id = require_ident("principal_id", principal_id)
         base = self.resolve_channel(channel, channel_account_id)
-        with self._connect() as db:
+        with closing(self._connect()) as connection, connection as db:
             principal = db.execute(
                 "SELECT tenant_id, cell_id FROM principal_routes WHERE "
                 "channel=? AND channel_account_id=? AND principal_id=?",

active-runtime/app/aios_v2r1/runtime.py

@@ -0,0 +1,136 @@
+"""V2R1 runtime composition root for authenticated channel ingress."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime
+from pathlib import Path
+from typing import Any, Callable, Mapping
+
+from .channel_adapter import ChannelAdapter, IngressResult
+from .identity import (
+    AuthenticationError,
+    IngressAuthenticator,
+    RuntimeIdentity,
+    RuntimeIdentityResolver,
+)
+from .message_ledger import MessageLedger
+from .router import TenantRouter
+from .session_pool import SessionPool
+from .tenant_context import ConflictError, ScopeError, TenantContext, ValidationError, utc_now
+
+
+@dataclass(frozen=True)
+class RuntimeIngressResponse:
+    """Sanitised boundary response; no exception text crosses external ingress."""
+
+    status: str
+    result: IngressResult | None = None
+
+
+class RuntimeWiring:
+    """Own the only external-to-runtime identity binding and message path.
+
+    The request accepted by :meth:`ingest` is an HMAC-authenticated envelope;
+    tenant, cell, channel, account, session, principal, and message claims are
+    signed together.  The envelope is then resolved against ``TenantRouter``.
+    Only that resolved identity can mint the short-lived internal context passed
+    to ``ChannelAdapter``.
+    """
+
+    def __init__(
+        self,
+        *,
+        router: TenantRouter,
+        ledger: MessageLedger,
+        session_pool: SessionPool,
+        channel_adapter: ChannelAdapter,
+        identity_resolver: RuntimeIdentityResolver,
+        ingress_authenticator: IngressAuthenticator,
+        internal_signing_key: bytes,
+        clock: Callable[[], datetime],
+    ) -> None:
+        self.router = router
+        self.ledger = ledger
+        self.session_pool = session_pool
+        self.channel_adapter = channel_adapter
+        self.identity_resolver = identity_resolver
+        self._ingress_authenticator = ingress_authenticator
+        self._internal_signing_key = internal_signing_key
+        self._clock = clock
+
+    @classmethod
+    def create(
+        cls,
+        root: str | Path,
+        *,
+        signing_key: bytes,
+        ingress_hmac_key: bytes | None = None,
+        clock: Callable[[], datetime] = utc_now,
+    ) -> "RuntimeWiring":
+        if not isinstance(signing_key, bytes) or len(signing_key) < 32:
+            raise ValidationError("internal signing key must be at least 32 bytes")
+        # Kept optional for one controlled migration release; the external
+        # request path still cannot supply a TenantContext token.
+        hmac_key = ingress_hmac_key if ingress_hmac_key is not None else signing_key
+        root_path = Path(root)
+        root_path.mkdir(parents=True, exist_ok=True)
+        router = TenantRouter(root_path / "router.sqlite3", clock=clock)
+        ledger = MessageLedger(root_path / "message_ledger.sqlite3")
+        session_pool = SessionPool(router)
+        adapter = ChannelAdapter(
+            router=router,
+            ledger=ledger,
+            session_pool=session_pool,
+            signing_key=signing_key,
+            clock=clock,
+        )
+        return cls(
+            router=router,
+            ledger=ledger,
+            session_pool=session_pool,
+            channel_adapter=adapter,
+            identity_resolver=RuntimeIdentityResolver(router),
+            ingress_authenticator=IngressAuthenticator(hmac_key),
+            internal_signing_key=signing_key,
+            clock=clock,
+        )
+
+    def _issue_internal_context(self, identity: RuntimeIdentity) -> TenantContext:
+        """Mint internal scope only after all claims matched the router route."""
+        return TenantContext.issue(
+            lifetime_seconds=300,
+            now=self._clock(),
+            tenant_id=identity.tenant_id,
+            cell_id=identity.cell_id,
+            principal_id=identity.principal_id,
+            principal_type="authenticated_user",
+            session_id=identity.session_id,
+            task_id=None,
+            channel=identity.channel,
+            channel_account_id=identity.channel_account_id,
+        )
+
+    def ingest(self, *, request: Mapping[str, Any]) -> RuntimeIngressResponse:
+        """Authenticate, resolve, issue internal context, then append and lease.
+
+        The only successful route is ``IngressAuthenticator`` ->
+        ``RuntimeIdentityResolver`` -> ``ChannelAdapter`` ->
+        ``MessageLedger.append`` + ``SessionPool``.  The legacy caller-supplied
+        TenantContext token interface is intentionally not accepted.
+        """
+        try:
+            ingress = self._ingress_authenticator.authenticate(request)
+            identity = self.identity_resolver.resolve(ingress.principal, ingress.claims)
+            context = self._issue_internal_context(identity)
+            result = self.channel_adapter.ingest(
+                token=context.sign(self._internal_signing_key), message=ingress.message
+            )
+            return RuntimeIngressResponse(status="ACCEPTED", result=result)
+        except (AuthenticationError, ScopeError):
+            return RuntimeIngressResponse(status="AUTH_FAILED")
+        except ValidationError:
+            return RuntimeIngressResponse(status="VALIDATION_FAILED")
+        except ConflictError:
+            return RuntimeIngressResponse(status="CONFLICT")
+        except Exception:
+            return RuntimeIngressResponse(status="INTERNAL_ERROR")

active-runtime/app/aios_v2r1/session_pool.py

@@ -0,0 +1,59 @@
+"""Process-local session lease pool backed by TenantRouter authority."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from threading import RLock
+
+from .router import Route, TenantRouter
+from .tenant_context import ScopeError, TenantContext
+
+
+@dataclass(frozen=True)
+class SessionLease:
+    tenant_id: str
+    cell_id: str
+    principal_id: str
+    session_id: str
+
+
+class SessionPool:
+    """Tracks active leases without creating a second session authority store."""
+
+    def __init__(self, router: TenantRouter) -> None:
+        self._router = router
+        self._leases: dict[tuple[str, str], SessionLease] = {}
+        self._lock = RLock()
+
+    @staticmethod
+    def _assert_route(context: TenantContext, route: Route) -> None:
+        if (
+            route.tenant_id != context.tenant_id
+            or route.cell_id != context.cell_id
+            or route.principal_id != context.principal_id
+            or route.session_id != context.session_id
+        ):
+            raise ScopeError("session lease does not match authoritative route")
+
+    def assert_available(self, context: TenantContext, route: Route) -> None:
+        """Re-resolve the context from the router before a ledger mutation."""
+        self._assert_route(context, route)
+        self._assert_route(context, self._router.assert_context_route(context))
+
+    def acquire(self, context: TenantContext, route: Route) -> SessionLease:
+        self._assert_route(context, route)
+        lease = SessionLease(
+            tenant_id=context.tenant_id,
+            cell_id=context.cell_id,
+            principal_id=context.principal_id,
+            session_id=context.session_id,
+        )
+        with self._lock:
+            return self._leases.setdefault((lease.tenant_id, lease.session_id), lease)
+
+    def release(self, lease: SessionLease) -> None:
+        with self._lock:
+            self._leases.pop((lease.tenant_id, lease.session_id), None)
+
+    def active_count(self) -> int:
+        with self._lock:
+            return len(self._leases)

agents/registry.yaml

@@ -1,8 +1,7 @@
-<<<<<<< HEAD
 # AIOS Agent Command Plane Registry
 # Git is the single source of truth for all agent commands.
 version: 1.0.0
-created_at: 2026-07-08T15:20:00+08:00
+created_at: "2026-07-08T15:20:00+08:00"
 baseline: V1R3C1
 
 commands:
@@ -12,6 +11,22 @@ commands:
   completed_path: "agents/commands/completed/"
   failed_path: "agents/commands/failed/"
 
+directories:
+  commands/active: "New commands pending execution — Git is the only entry point"
+  commands/running: "Commands currently being executed by an agent"
+  commands/completed: "Commands that finished successfully — moved here on success"
+  commands/failed: "Commands that finished with errors — moved here on failure"
+
+lifecycle:
+  new: active/
+  running: running/
+  success: completed/
+  failure: failed/
+
+# Command lifecycle state machine:
+# active/ → locked & moved to running/ → execute → success → completed/
+#                                                          → failure → failed/
+
 policies:
   path: "agents/policies/"
   source: "git"
@@ -38,30 +53,4 @@ audit:
   agent_runs: "audit/agent-runs/"
   version_history: "audit/version-history/"
   command_history: "audit/command-history/"
-=======
-# AIOS Agent Command Registry
-# Git is the single source of truth for all agent commands.
-
-registry_version: 1.0.0
-created_at: "2026-07-08T15:44:00+08:00"
-baseline: V1R3C1
-
-directories:
-  commands/active:    "New commands pending execution — Git is the only entry point"
-  commands/running:   "Commands currently being executed by an agent"
-  commands/completed: "Commands that finished successfully — moved here on success"
-  commands/failed:    "Commands that finished with errors — moved here on failure"
-
-state: agents/state/
-policies: agents/policies/
-
-lifecycle:
-  new:     active/
-  running: running/
-  success: completed/
-  failure: failed/
-
-# Command lifecycle state machine:
-# active/ → locked & moved to running/ → execute → success → completed/
-#                                                          → failure → failed/
->>>>>>> 4d18ee0 (feat(cmd): enable Agent Command Plane, audit plane, and Web UI dashboard)
+  interaction_history: "audit/interaction-history/"

docs/architecture/AIOS_AGENT_AUDIT_PLANE.md

@@ -1,4 +1,3 @@
-<<<<<<< HEAD
 # AIOS Agent Audit Plane Architecture
 
 ## Overview
@@ -25,9 +24,10 @@ Agent Command Plane (agents/commands/)
     +---> Audit
     |         |
     |         v
-    |     audit/agent-runs/        (per-agent log)
-    |     audit/command-history/   (per-command log)
-    |     audit/version-history/   (per-baseline log)
+    |     audit/agent-runs/          (per-agent log)
+    |     audit/command-history/     (per-command log)
+    |     audit/version-history/     (per-baseline log)
+    |     audit/interaction-history/ (agent conversation traces)
     |
     +---> Git Commit
               |
@@ -45,8 +45,39 @@ Agent Command Plane (agents/commands/)
 5. Git commit locks the audit record
 6. Web UI displays timeline from git history
 
+## Directory Structure
+
+```
+audit/
+├── agent-runs/          # Per-execution records (JSON)
+├── command-history/     # Command lifecycle history
+├── version-history/     # Version/baseline change log
+└── interaction-history/ # Agent conversation traces
+```
+
 ## Audit Record Schema
 
+Current agent-run records use the following fields:
+
+| Field | Description |
+|-------|-------------|
+| agent_run_id | Unique execution ID |
+| command | Command file name |
+| agent | Executing agent name |
+| version | Baseline version at execution time |
+| objective | Human-readable goal |
+| start_time | ISO-8601 start timestamp |
+| end_time | ISO-8601 completion timestamp |
+| interaction_count | Agent-owner interaction turns |
+| tool_call_count | Tool invocations during execution |
+| changed_files | Files modified in this run |
+| commit_sha | Git commit hash |
+| test_result | Test verification outcome |
+| final_result | SUCCESS / FAILED / PARTIAL |
+| evidence_path | Link to evidence directory |
+
+Legacy records remain valid and may use the following compatibility shape:
+
 ```json
 {
   "id": "",
@@ -72,51 +103,6 @@ Agent Command Plane (agents/commands/)
 - Every code change must be traceable to who/what agent created it
 - No agent-created changes without audit trail
 
-## Related
-
-- [Agent Command Plane](../agents/README.md)
-- [Baseline System](baseline-system.md)
-=======
-# AIOS Agent Audit Plane
-
-## Purpose
-
-Every agent action is recorded. All code changes are traceable to:
-- Who made the change
-- Why the change was made
-- When it was made
-- Which version introduced it
-- What test results showed
-
-## Directory Structure
-
-```
-audit/
-├── agent-runs/         # Per-execution records (JSON)
-├── command-history/    # Command lifecycle history
-├── version-history/    # Version/baseline change log
-└── interaction-history/ # Agent conversation traces
-```
-
-## Agent Run Record
-
-| Field | Description |
-|-------|-------------|
-| agent_run_id | Unique execution ID |
-| command | Command file name |
-| agent | Executing agent name |
-| version | Baseline version at execution time |
-| objective | Human-readable goal |
-| start_time | ISO-8601 start timestamp |
-| end_time | ISO-8601 completion timestamp |
-| interaction_count | Agent-owner interaction turns |
-| tool_call_count | Tool invocations during execution |
-| changed_files | Files modified in this run |
-| commit_sha | Git commit hash |
-| test_result | Test verification outcome |
-| final_result | SUCCESS / FAILED / PARTIAL |
-| evidence_path | Link to evidence directory |
-
 ## Version History
 
 Records baseline promotions, rollbacks, and version changes.
@@ -127,6 +113,9 @@ Conversation logs between agent and owner for each command execution.
 
 ## Audit Trail Requirement
 
-All changes tracked via Git with signed commits where possible.
-Every commit message references the originating Command ID.
->>>>>>> 4d18ee0 (feat(cmd): enable Agent Command Plane, audit plane, and Web UI dashboard)
+All changes are tracked via Git with signed commits where possible. Every commit references the originating Command ID.
+
+## Related
+
+- [Agent Command Plane](../agents/README.md)
+- [Baseline System](baseline-system.md)

tests/integration/test_runtime_ingress_flow.py

@@ -0,0 +1,143 @@
+"""End-to-end boundary tests for the V2R1 authenticated ingress runtime."""
+from __future__ import annotations
+
+import hashlib
+import hmac
+import sys
+from datetime import datetime, timezone
+from tempfile import TemporaryDirectory
+
+from aios_v2r1.runtime import RuntimeWiring
+from aios_v2r1.message_ledger import MessageLedger
+from aios_v2r1.tenant_context import canonical_json
+
+
+class FixedClock:
+    def __init__(self) -> None:
+        self.now = datetime(2026, 7, 13, tzinfo=timezone.utc)
+
+    def __call__(self) -> datetime:
+        return self.now
+
+
+class TestRuntimeIngressFlow:
+    """Use the public ingress boundary; do not reach into router internals."""
+
+    internal_key = b"runtime-integration-internal-key-32bytes"
+    ingress_key = b"runtime-integration-ingress-key-32bytes!"
+
+    def setup_method(self) -> None:
+        self.directory = TemporaryDirectory()
+        self.clock = FixedClock()
+        self.runtime = RuntimeWiring.create(
+            self.directory.name,
+            signing_key=self.internal_key,
+            ingress_hmac_key=self.ingress_key,
+            clock=self.clock,
+        )
+        self.runtime.router.register_tenant("tenant-a", "cell-a")
+        self.runtime.router.register_tenant("tenant-b", "cell-b")
+        self.runtime.router.bind_channel(
+            tenant_id="tenant-a", channel="weixin", channel_account_id="account-a"
+        )
+        self.runtime.router.bind_channel(
+            tenant_id="tenant-b", channel="weixin", channel_account_id="account-b"
+        )
+        self.runtime.router.bind_principal(
+            tenant_id="tenant-a",
+            channel="weixin",
+            channel_account_id="account-a",
+            principal_id="user-a",
+        )
+        self.runtime.router.bind_session(
+            tenant_id="tenant-a",
+            session_id="session-a",
+            cell_id="cell-a",
+            principal_id="user-a",
+            ttl_seconds=60,
+        )
+
+    def teardown_method(self) -> None:
+        self.directory.cleanup()
+
+    def _message(self, *, event_id: str = "event-a", message_id: str = "message-a",
+                 body: str = "hello", **changes: object) -> dict[str, object]:
+        values: dict[str, object] = {
+            "tenant_id": "tenant-a", "cell_id": "cell-a", "principal_id": "user-a",
+            "principal_type": "authenticated_user", "origin_type": "channel_user",
+            "channel": "weixin", "channel_account_id": "account-a", "peer_id": "peer-a",
+            "conversation_id": "conversation-a", "session_id": "session-a", "task_id": None,
+            "event_id": event_id, "message_id": message_id, "reply_to_message_id": None,
+            "direction": "inbound", "command_capable": True,
+            "occurred_at": "2026-07-13T00:00:00+00:00",
+            "received_at": "2026-07-13T00:00:01+00:00",
+            "content": {"body": body, "quoted_body": None},
+            "adapter": {"name": "weixin", "version": "v2"},
+        }
+        values.update(changes)
+        return values
+
+    def _request(self, **changes: object) -> dict[str, object]:
+        values: dict[str, object] = {
+            "principal_id": "user-a", "tenant_id": "tenant-a", "cell_id": "cell-a",
+            "channel": "weixin", "channel_account_id": "account-a", "session_id": "session-a",
+            "message": self._message(),
+        }
+        values.update(changes)
+        values["signature"] = hmac.new(
+            self.ingress_key, canonical_json(values), hashlib.sha256
+        ).hexdigest()
+        return values
+
+    def test_authenticated_ingress_uses_direct_append_and_acquires_session(self) -> None:
+        # Observe the real MessageLedger object while preserving every method
+        # implementation.  This proves the public ingress path never calls
+        # the legacy compatibility entrypoint; it is not a mock or bypass.
+        remember_code = MessageLedger.remember.__code__
+        remember_calls = 0
+        previous_profile = sys.getprofile()
+
+        def observe(frame: object, event: str, arg: object) -> None:
+            nonlocal remember_calls
+            if event == "call" and getattr(frame, "f_code", None) is remember_code:
+                remember_calls += 1
+
+        sys.setprofile(observe)
+        try:
+            response = self.runtime.ingest(request=self._request())
+        finally:
+            sys.setprofile(previous_profile)
+
+        assert response.status == "ACCEPTED"
+        assert response.result is not None
+        assert response.result.record.message_id == "message-a"
+        assert response.result.session.session_id == "session-a"
+        assert self.runtime.ledger.count() == 1
+        assert self.runtime.session_pool.active_count() == 1
+        assert remember_calls == 0
+
+    def test_forged_tenant_claim_is_rejected_before_ledger_or_session_mutation(self) -> None:
+        response = self.runtime.ingest(request=self._request(
+            tenant_id="tenant-b",
+            cell_id="cell-b",
+            message=self._message(tenant_id="tenant-b", cell_id="cell-b"),
+        ))
+
+        assert response.status == "AUTH_FAILED"
+        assert response.result is None
+        assert self.runtime.ledger.count() == 0
+        assert self.runtime.session_pool.active_count() == 0
+
+    def test_conflicting_duplicate_message_is_mapped_at_the_ingress_boundary(self) -> None:
+        assert self.runtime.ingest(request=self._request(
+            message=self._message(body="first")
+        )).status == "ACCEPTED"
+
+        response = self.runtime.ingest(request=self._request(
+            message=self._message(event_id="event-b", body="second")
+        ))
+
+        assert response.status == "CONFLICT"
+        assert response.result is None
+        assert self.runtime.ledger.count() == 1
+        assert self.runtime.session_pool.active_count() == 1

tests/v2r1/test_message_ledger_append.py

@@ -0,0 +1,73 @@
+"""Contract tests for the runtime-facing direct MessageLedger append path."""
+from __future__ import annotations
+
+from concurrent.futures import ThreadPoolExecutor
+from tempfile import TemporaryDirectory
+import unittest
+
+from aios_v2r1.message_ledger import MessageLedger
+from aios_v2r1.tenant_context import CanonicalMessage, ConflictError
+
+
+def message(*, tenant_id: str = "tenant-a", cell_id: str = "cell-a",
+            event_id: str = "event-a", message_id: str = "message-a",
+            body: str = "hello", reply_to_message_id: str | None = None) -> CanonicalMessage:
+    return CanonicalMessage(
+        tenant_id=tenant_id, cell_id=cell_id, principal_id="user-a",
+        principal_type="authenticated_user", origin_type="channel_user",
+        channel="weixin", channel_account_id="account-a", peer_id="peer-a",
+        conversation_id="conversation-a", session_id="session-a", task_id=None,
+        event_id=event_id, message_id=message_id, reply_to_message_id=reply_to_message_id,
+        direction="inbound", command_capable=True,
+        occurred_at="2026-07-13T00:00:00+00:00",
+        received_at="2026-07-13T00:00:01+00:00", body=body,
+        quoted_body=None, adapter_name="weixin", adapter_version="v2",
+    )
+
+
+class MessageLedgerAppendContractTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.directory = TemporaryDirectory()
+        self.addCleanup(self.directory.cleanup)
+        self.ledger = MessageLedger(f"{self.directory.name}/ledger.sqlite3")
+
+    def test_append_writes_directly_without_using_remember(self) -> None:
+        # A compatibility wrapper may remain, but the runtime append path may not
+        # route through it: this catches append() -> remember() regressions.
+        self.ledger.remember = lambda _: (_ for _ in ()).throw(AssertionError("remember called"))  # type: ignore[method-assign]
+        record = self.ledger.append(message())
+        self.assertEqual("message-a", record.message_id)
+        self.assertEqual(1, self.ledger.count())
+
+    def test_append_rejects_same_message_id_with_different_content(self) -> None:
+        self.ledger.append(message(body="first"))
+        with self.assertRaises(ConflictError):
+            self.ledger.append(message(event_id="event-b", body="second"))
+        self.assertEqual(1, self.ledger.count())
+
+    def test_append_is_idempotent_when_only_event_id_changes(self) -> None:
+        stored = self.ledger.append(message(event_id="event-a", body="same body"))
+        replay = self.ledger.append(message(event_id="event-b", body="same body"))
+
+        self.assertEqual(stored, replay)
+        self.assertEqual("event-a", replay.event_id)
+        self.assertEqual(1, self.ledger.count())
+
+    def test_append_keeps_tenants_isolated(self) -> None:
+        self.ledger.append(message())
+        self.ledger.append(message(tenant_id="tenant-b", cell_id="cell-b"))
+        self.assertEqual(2, self.ledger.count())
+        self.assertEqual("hello", self.ledger.resolve_quote(
+            message(tenant_id="tenant-b", cell_id="cell-b", event_id="event-c",
+                    message_id="message-c", body="reply", reply_to_message_id="message-a")
+        ))
+
+    def test_concurrent_identical_append_is_one_record(self) -> None:
+        with ThreadPoolExecutor(max_workers=16) as executor:
+            records = list(executor.map(lambda _: self.ledger.append(message()), range(16)))
+        self.assertEqual(1, self.ledger.count())
+        self.assertEqual({"message-a"}, {record.message_id for record in records})
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/v2r1/test_runtime_identity_binding.py

@@ -0,0 +1,159 @@
+"""Security contracts for authenticated OpenClaw runtime identity binding."""
+from __future__ import annotations
+
+import hashlib
+import hmac
+from datetime import datetime, timezone
+from tempfile import TemporaryDirectory
+import unittest
+
+from aios_v2r1.runtime import RuntimeWiring
+from aios_v2r1.tenant_context import canonical_json
+
+
+class MutableClock:
+    def __init__(self, now: datetime) -> None:
+        self.now = now
+
+    def __call__(self) -> datetime:
+        return self.now
+
+
+class RuntimeIdentityBindingTests(unittest.TestCase):
+    internal_key = b"internal-context-signing-key-is-32-bytes"
+    ingress_key = b"external-ingress-hmac-key-is-32-bytes"
+
+    def setUp(self) -> None:
+        self.directory = TemporaryDirectory()
+        self.addCleanup(self.directory.cleanup)
+        self.clock = MutableClock(datetime(2026, 7, 13, tzinfo=timezone.utc))
+        self.runtime = RuntimeWiring.create(
+            self.directory.name,
+            signing_key=self.internal_key,
+            ingress_hmac_key=self.ingress_key,
+            clock=self.clock,
+        )
+        self.runtime.router.register_tenant("tenant-a", "cell-a")
+        self.runtime.router.register_tenant("tenant-b", "cell-b")
+        self.runtime.router.bind_channel(
+            tenant_id="tenant-a", channel="weixin", channel_account_id="account-a"
+        )
+        self.runtime.router.bind_channel(
+            tenant_id="tenant-b", channel="weixin", channel_account_id="account-b"
+        )
+        self.runtime.router.bind_principal(
+            tenant_id="tenant-a", channel="weixin", channel_account_id="account-a",
+            principal_id="user-a",
+        )
+        self.runtime.router.bind_session(
+            tenant_id="tenant-a", session_id="session-a", cell_id="cell-a",
+            principal_id="user-a", ttl_seconds=60,
+        )
+
+    def message(self, **changes: object) -> dict[str, object]:
+        values: dict[str, object] = {
+            "tenant_id": "tenant-a", "cell_id": "cell-a", "principal_id": "user-a",
+            "principal_type": "authenticated_user", "origin_type": "channel_user",
+            "channel": "weixin", "channel_account_id": "account-a", "peer_id": "peer-a",
+            "conversation_id": "conversation-a", "session_id": "session-a", "task_id": None,
+            "event_id": "event-a", "message_id": "message-a", "reply_to_message_id": None,
+            "direction": "inbound", "command_capable": True,
+            "occurred_at": "2026-07-13T00:00:00+00:00",
+            "received_at": "2026-07-13T00:00:01+00:00",
+            "content": {"body": "hello", "quoted_body": None},
+            "adapter": {"name": "weixin", "version": "v2"},
+        }
+        values.update(changes)
+        return values
+
+    def authenticated_principal_type(self):
+        try:
+            from aios_v2r1.identity import AuthenticatedPrincipal
+        except ImportError as exc:
+            self.fail(f"Identity resolver contract is missing: {exc}")
+        return AuthenticatedPrincipal
+
+    def signed_request(self, **changes: object) -> dict[str, object]:
+        values: dict[str, object] = {
+            "principal_id": "user-a",
+            "tenant_id": "tenant-a",
+            "cell_id": "cell-a",
+            "channel": "weixin",
+            "channel_account_id": "account-a",
+            "session_id": "session-a",
+            "message": self.message(),
+        }
+        values.update(changes)
+        signed = canonical_json(values)
+        values["signature"] = hmac.new(
+            self.ingress_key, signed, hashlib.sha256
+        ).hexdigest()
+        return values
+
+    def test_raw_payload_cannot_construct_authenticated_principal(self) -> None:
+        AuthenticatedPrincipal = self.authenticated_principal_type()
+        with self.assertRaises(TypeError):
+            AuthenticatedPrincipal("user-a")  # type: ignore[call-arg]
+
+    def test_forged_tenant_claim_is_rejected_after_valid_hmac(self) -> None:
+        request = self.signed_request(
+            tenant_id="tenant-b", cell_id="cell-b",
+            message=self.message(tenant_id="tenant-b", cell_id="cell-b"),
+        )
+
+        result = self.runtime.ingest(request=request)
+
+        self.assertEqual("AUTH_FAILED", result.status)
+        self.assertIsNone(result.result)
+        self.assertEqual(0, self.runtime.ledger.count())
+
+    def test_invalid_principal_is_rejected_after_valid_hmac(self) -> None:
+        request = self.signed_request(
+            principal_id="unknown-user",
+            message=self.message(principal_id="unknown-user"),
+        )
+
+        result = self.runtime.ingest(request=request)
+
+        self.assertEqual("AUTH_FAILED", result.status)
+        self.assertIsNone(result.result)
+        self.assertEqual(0, self.runtime.ledger.count())
+
+    def test_registered_principal_is_resolved_before_internal_context_is_issued(self) -> None:
+        request = self.signed_request()
+        result = self.runtime.ingest(request=request)
+
+        self.assertEqual("ACCEPTED", result.status)
+        self.assertIsNotNone(result.result)
+        assert result.result is not None
+        self.assertEqual("tenant-a", result.result.route.tenant_id)
+        self.assertEqual("message-a", result.result.record.message_id)
+        self.assertEqual(1, self.runtime.ledger.count())
+        self.assertEqual(1, self.runtime.session_pool.active_count())
+
+    def test_cross_tenant_channel_claim_cannot_retarget_authenticated_principal(self) -> None:
+        request = self.signed_request(
+            tenant_id="tenant-b", cell_id="cell-b", channel_account_id="account-b",
+            message=self.message(
+                tenant_id="tenant-b", cell_id="cell-b", channel_account_id="account-b"
+            ),
+        )
+
+        result = self.runtime.ingest(request=request)
+
+        self.assertEqual("AUTH_FAILED", result.status)
+        self.assertEqual(0, self.runtime.ledger.count())
+        self.assertEqual(0, self.runtime.session_pool.active_count())
+
+    def test_hmac_binds_all_scope_claims(self) -> None:
+        request = self.signed_request()
+        request["tenant_id"] = "tenant-b"
+
+        result = self.runtime.ingest(request=request)
+
+        self.assertEqual("AUTH_FAILED", result.status)
+        self.assertEqual(0, self.runtime.ledger.count())
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/v2r1/test_runtime_wiring.py

@@ -0,0 +1,154 @@
+"""Integration contracts for the V2R1 authenticated runtime composition root."""
+from __future__ import annotations
+
+from concurrent.futures import ThreadPoolExecutor
+from datetime import datetime, timedelta, timezone
+import hashlib
+import hmac
+from tempfile import TemporaryDirectory
+import unittest
+
+from aios_v2r1.runtime import RuntimeWiring
+from aios_v2r1.tenant_context import canonical_json
+
+
+class MutableClock:
+    def __init__(self, now: datetime) -> None:
+        self.now = now
+
+    def __call__(self) -> datetime:
+        return self.now
+
+
+class RuntimeWiringIntegrationTests(unittest.TestCase):
+    internal_key = b"runtime-internal-context-key-is-32bytes"
+    ingress_key = b"runtime-external-ingress-key-is-32bytes"
+
+    def setUp(self) -> None:
+        self.directory = TemporaryDirectory()
+        self.addCleanup(self.directory.cleanup)
+        self.clock = MutableClock(datetime(2026, 7, 13, tzinfo=timezone.utc))
+        self.runtime = RuntimeWiring.create(
+            self.directory.name,
+            signing_key=self.internal_key,
+            ingress_hmac_key=self.ingress_key,
+            clock=self.clock,
+        )
+        self.runtime.router.register_tenant("tenant-a", "cell-a")
+        self.runtime.router.register_tenant("tenant-b", "cell-b")
+        self.runtime.router.bind_channel(
+            tenant_id="tenant-a", channel="weixin", channel_account_id="account-a"
+        )
+        self.runtime.router.bind_channel(
+            tenant_id="tenant-b", channel="weixin", channel_account_id="account-b"
+        )
+        self.runtime.router.bind_principal(
+            tenant_id="tenant-a", channel="weixin", channel_account_id="account-a",
+            principal_id="user-a",
+        )
+        self.runtime.router.bind_session(
+            tenant_id="tenant-a", session_id="session-a", cell_id="cell-a",
+            principal_id="user-a", ttl_seconds=60,
+        )
+
+    def message(
+        self, *, event_id: str = "event-a", message_id: str = "message-a",
+        body: str = "hello", **changes: object,
+    ) -> dict[str, object]:
+        values: dict[str, object] = {
+            "tenant_id": "tenant-a", "cell_id": "cell-a", "principal_id": "user-a",
+            "principal_type": "authenticated_user", "origin_type": "channel_user",
+            "channel": "weixin", "channel_account_id": "account-a", "peer_id": "peer-a",
+            "conversation_id": "conversation-a", "session_id": "session-a", "task_id": None,
+            "event_id": event_id, "message_id": message_id, "reply_to_message_id": None,
+            "direction": "inbound", "command_capable": True,
+            "occurred_at": "2026-07-13T00:00:00+00:00",
+            "received_at": "2026-07-13T00:00:01+00:00",
+            "content": {"body": body, "quoted_body": None},
+            "adapter": {"name": "weixin", "version": "v2"},
+        }
+        values.update(changes)
+        return values
+
+    def request(self, **changes: object) -> dict[str, object]:
+        values: dict[str, object] = {
+            "principal_id": "user-a", "tenant_id": "tenant-a", "cell_id": "cell-a",
+            "channel": "weixin", "channel_account_id": "account-a", "session_id": "session-a",
+            "message": self.message(),
+        }
+        values.update(changes)
+        values["signature"] = hmac.new(
+            self.ingress_key, canonical_json(values), hashlib.sha256
+        ).hexdigest()
+        return values
+
+    def ingest(self, **changes: object):
+        return self.runtime.ingest(request=self.request(**changes))
+
+    def test_real_ingress_appends_exact_record_and_reopens(self) -> None:
+        first = self.ingest()
+        self.assertEqual("ACCEPTED", first.status)
+        assert first.result is not None
+        self.assertEqual("message-a", first.result.record.message_id)
+        self.assertEqual("tenant-a", first.result.route.tenant_id)
+        self.assertEqual(1, self.runtime.ledger.count())
+        self.assertEqual(1, self.runtime.session_pool.active_count())
+
+        reopened = RuntimeWiring.create(
+            self.directory.name, signing_key=self.internal_key,
+            ingress_hmac_key=self.ingress_key, clock=self.clock,
+        )
+        repeated = reopened.ingest(request=self.request())
+        self.assertEqual("ACCEPTED", repeated.status)
+        assert repeated.result is not None
+        self.assertEqual(first.result.record, repeated.result.record)
+        self.assertEqual(1, reopened.ledger.count())
+        self.assertEqual(1, reopened.session_pool.active_count())
+
+    def test_authoritative_tenant_route_rejects_cross_tenant_ingress_without_mutation(self) -> None:
+        result = self.ingest(
+            tenant_id="tenant-b", cell_id="cell-b",
+            message=self.message(tenant_id="tenant-b", cell_id="cell-b"),
+        )
+        self.assertEqual("AUTH_FAILED", result.status)
+        self.assertIsNone(result.result)
+        self.assertEqual(0, self.runtime.ledger.count())
+        self.assertEqual(0, self.runtime.session_pool.active_count())
+
+    def test_identical_concurrent_ingress_is_one_exact_ledger_record_and_one_session(self) -> None:
+        with ThreadPoolExecutor(max_workers=24) as executor:
+            results = list(executor.map(lambda _: self.ingest(), range(24)))
+        self.assertEqual({"ACCEPTED"}, {result.status for result in results})
+        self.assertEqual(1, self.runtime.ledger.count())
+        self.assertEqual(1, self.runtime.session_pool.active_count())
+
+    def test_collision_is_sanitized_and_does_not_leave_partial_session_or_ledger_state(self) -> None:
+        self.assertEqual("ACCEPTED", self.ingest(
+            message=self.message(event_id="event-a", message_id="message-a", body="first")
+        ).status)
+        collision = self.ingest(
+            message=self.message(event_id="event-a", message_id="message-b", body="second")
+        )
+        self.assertEqual("CONFLICT", collision.status)
+        self.assertIsNone(collision.result)
+        self.assertEqual(1, self.runtime.ledger.count())
+        self.assertEqual(1, self.runtime.session_pool.active_count())
+
+    def test_unknown_cross_and_expired_sessions_fail_closed_without_mutation(self) -> None:
+        unknown = self.ingest(
+            session_id="unknown-session", message=self.message(session_id="unknown-session")
+        )
+        self.assertEqual("AUTH_FAILED", unknown.status)
+        cross = self.ingest(
+            principal_id="user-b", message=self.message(principal_id="user-b")
+        )
+        self.assertEqual("AUTH_FAILED", cross.status)
+        self.clock.now += timedelta(seconds=61)
+        expired = self.ingest()
+        self.assertEqual("AUTH_FAILED", expired.status)
+        self.assertEqual(0, self.runtime.ledger.count())
+        self.assertEqual(0, self.runtime.session_pool.active_count())
+
+
+if __name__ == "__main__":
+    unittest.main()