Add OpenClaw deployment command for Agent GPT message plane

Add Agent GPT message plane configuration
Add Agent GPT message plane admin dashboard model
2026-07-08 16:33:46 +08:00 · 2026-07-08 16:33:39 +08:00 · 2026-07-08 16:33:34 +08:00 · 2026-07-08 16:33:30 +08:00 · 2026-07-08 16:33:25 +08:00 · 2026-07-08 16:33:16 +08:00
248 changed files with 431 additions and 14044 deletions
@@ -1,38 +0,0 @@
 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.
@@ -1,95 +0,0 @@
 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.
@@ -1,131 +0,0 @@
 ---
 status: PASS
 commit: 502ef30642597fb0b70b90ae2e8a0b6af217d80e
 task_id: TASK-CODEX-V2R1-G1-FINAL-VERIFY-RETRY-20260715T010807UTC8
 executor: CODEX_CANDIDATE_BUILDER
 source_commit: 502ef30642597fb0b70b90ae2e8a0b6af217d80e
 source_branch: work/v2r1-integration-quality-2000
 candidate_ref: work/codex/v2r1-g1-recovery-candidate-20260715T010807UTC8
 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
 commands:
  - id: v2r1_tests
    exit_code: 0
  - id: integration_tests
    exit_code: 0
  - id: conflict_object_scan
    command: >-
      git grep -n -E '^(<<<<<<<|=======|>>>>>>>)'
      502ef30642597fb0b70b90ae2e8a0b6af217d80e --
      . ':(exclude)source-baseline/*.zip'
    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: webops_ignore_order
    command: >-
      grep -E '^/data/webops\.sqlite3(-wal|-shm|-journal)?$' .gitignore
    exit_code: 0
  - id: gitleaks_version
    command: /tmp/gitleaks-codex-8.30.1/gitleaks version
    exit_code: 0
  - id: gitleaks_git
    command: >-
      /tmp/gitleaks-codex-8.30.1/gitleaks git --no-banner --no-color
      --redact --report-format json --report-path "$TMP/git.json"
      --log-opts=502ef30642597fb0b70b90ae2e8a0b6af217d80e
      /data/aios-gitops/worktrees/chatgpt/TASK-CODEX-V2R1-G1-FINAL-VERIFY-RETRY-20260715T010807UTC8
    exit_code: 0
  - id: gitleaks_dir
    command: >-
      /tmp/gitleaks-codex-8.30.1/gitleaks dir --no-banner --no-color
      --redact --report-format json --report-path "$TMP/dir.json"
      /data/aios-gitops/worktrees/chatgpt/TASK-CODEX-V2R1-G1-FINAL-VERIFY-RETRY-20260715T010807UTC8
    exit_code: 0
 conflict_scan:
  status: PASS
  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
  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:
    exit_code: 0
    findings: 0
    report_json: []
  dir:
    exit_code: 0
    findings: 0
    report_json: []
 hmac:
  status: PASS
  method: AST assignment inspection
  target: tests/v2r1/test_runtime_wiring.py
  command: /usr/bin/python3 - tests/v2r1/test_runtime_wiring.py
  exit_code: 0
  internal_key_length: 39
  ingress_key_length: 39
 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 exact candidate commit to independent verification only; do not deploy, merge, tag, or alter baselines.
 ---
@@ -10,8 +10,3 @@ node_modules/
 runtime/tmp/
 runtime/cache/
 logs/
 config/test_write_perms
 /data/webops.sqlite3
 /data/webops.sqlite3-wal
 /data/webops.sqlite3-shm
 /data/webops.sqlite3-journal
@@ -1 +1 @@
-V1R4C6
+V1R1C4
@@ -1,56 +0,0 @@
 """API adapter for agent context routing and visibility."""
 from typing import Any
 try:
    from fastapi import APIRouter
 except Exception:
    APIRouter = None
 from .agent_context_dashboard import AgentContextDashboard
 from .agent_context_rbac import AgentContextRBAC
 from .agent_context_resolver import AgentContextResolver
 resolver = AgentContextResolver()
 rbac = AgentContextRBAC()
 dashboard = AgentContextDashboard()
 def resolve_agent_context(payload: dict[str, Any]) -> dict[str, Any]:
    route = rbac.route(payload["user_id"], payload.get("group_id"), payload.get("role", "user"))
    context = resolver.resolve(
        user_id=payload["user_id"],
        group_id=payload.get("group_id", ""),
        session=payload.get("session"),
        user_profile=payload.get("user_profile"),
        group_policy=payload.get("group_policy"),
        agent_default=payload.get("agent_default"),
        owner_policy=payload.get("owner_policy"),
        owner=payload.get("role") == "owner",
    )
    return {"routing": route.to_dict(), "context": context.to_dict()}
 def get_context_dashboard() -> dict[str, Any]:
    return dashboard.overview()
 def build_router():
    if APIRouter is None:
        return None
    router = APIRouter(prefix="/admin/ai-evolution/context", tags=["context-isolation"])
    @router.post("/resolve", operation_id="resolveAgentContext")
    def resolve(payload: dict[str, Any]):
        return resolve_agent_context(payload)
    @router.get("/overview", operation_id="getAgentContextDashboard")
    def overview():
        return get_context_dashboard()
    return router
 router = build_router()
 __all__ = ["router", "resolve_agent_context", "get_context_dashboard"]
@@ -1,23 +0,0 @@
 """Audit context resolution decisions."""
 from datetime import datetime, timezone
 from typing import Any
 def utc_now():
    return datetime.now(timezone.utc).isoformat()
 class AgentContextAudit:
    def record(self, context: dict[str, Any]) -> dict[str, Any]:
        return {
            "event": "context_loaded",
            "agent_id": context.get("agent_id"),
            "user_id": context.get("user_id"),
            "group_id": context.get("group_id"),
            "layers": [x.get("name") for x in context.get("layers", [])],
            "timestamp": utc_now(),
        }
 __all__ = ["AgentContextAudit"]
@@ -1,47 +0,0 @@
 """Web UI model for context isolation visibility."""
 from __future__ import annotations
 from typing import Any
 class AgentContextDashboard:
    def overview(self, contexts: list[dict[str, Any]] | None = None) -> dict[str, Any]:
        contexts = contexts or []
        return {
            "module": "agent-context-isolation",
            "route": "/admin/ai-evolution/context",
            "title": "Agent Context Isolation",
            "panels": [
                {
                    "name": "agent-routing",
                    "title": "Agent 路由",
                    "data": contexts,
                },
                {
                    "name": "context-layers",
                    "title": "Context 来源层级",
                    "layers": [
                        "safety_policy",
                        "owner_policy",
                        "group_policy",
                        "role_policy",
                        "user_profile",
                        "session_context",
                        "agent_default",
                    ],
                },
                {
                    "name": "audit",
                    "title": "Context 审计",
                    "events": [
                        "context_loaded",
                        "profile_updated",
                        "agent_selected",
                    ],
                },
            ],
        }
 __all__ = ["AgentContextDashboard"]
@@ -1,55 +0,0 @@
 """RBAC binding for group main agents and owner agents."""
 from __future__ import annotations
 from dataclasses import asdict, dataclass
 from typing import Any
@dataclass(frozen=True)
 class AgentRoutingDecision:
    user_id: str
    group_id: str
    role: str
    agent_id: str
    scope: str
    allowed: bool
    reason: str
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
 class AgentContextRBAC:
    """Routes users to group agents and owners to dedicated owner agents."""
    VALID_ROLES = {"user", "group_admin", "owner"}
    def route(self, user_id: str, group_id: str | None, role: str) -> AgentRoutingDecision:
        if role not in self.VALID_ROLES:
            return AgentRoutingDecision(user_id, group_id or "", role, "", "none", False, "unknown_role")
        if role == "owner":
            return AgentRoutingDecision(
                user_id=user_id,
                group_id=group_id or "global",
                role=role,
                agent_id=f"owner-agent:{user_id}",
                scope="global",
                allowed=True,
                reason="owner_dedicated_agent",
            )
        if not group_id:
            return AgentRoutingDecision(user_id, "", role, "", "none", False, "group_required")
        scope = "group" if role == "group_admin" else "self"
        return AgentRoutingDecision(
            user_id=user_id,
            group_id=group_id,
            role=role,
            agent_id=f"group-main-agent:{group_id}",
            scope=scope,
            allowed=True,
            reason="group_agent_routing",
        )
 __all__ = ["AgentRoutingDecision", "AgentContextRBAC"]
@@ -1,82 +0,0 @@
 """Multi-user agent context resolver.
 Architecture:
 - shared agent kernel
 - group main agent
 - owner dedicated agent
 - user profile overlay
 - session overlay
 Personal preferences never enter shared USER.md.
 """
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from typing import Any
 def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()
@dataclass(frozen=True)
 class ContextLayer:
    name: str
    priority: int
    data: dict[str, Any]
@dataclass
 class EffectiveContext:
    agent_id: str
    user_id: str
    group_id: str
    layers: list[ContextLayer]
    resolved: dict[str, Any]
    resolved_at: str = field(default_factory=utc_now)
    def to_dict(self):
        return {
            "agent_id": self.agent_id,
            "user_id": self.user_id,
            "group_id": self.group_id,
            "layers": [asdict(x) for x in self.layers],
            "resolved": self.resolved,
            "resolved_at": self.resolved_at,
        }
 class AgentContextResolver:
    """Resolves context without shared user-state pollution."""
    def resolve(
        self,
        user_id: str,
        group_id: str,
        session: dict[str, Any] | None = None,
        user_profile: dict[str, Any] | None = None,
        group_policy: dict[str, Any] | None = None,
        agent_default: dict[str, Any] | None = None,
        owner_policy: dict[str, Any] | None = None,
        owner: bool = False,
    ) -> EffectiveContext:
        layers = [
            ContextLayer("agent_default", 10, agent_default or {}),
            ContextLayer("group_policy", 30, group_policy or {}),
            ContextLayer("user_profile", 50, user_profile or {}),
            ContextLayer("session", 70, session or {}),
        ]
        agent_id = f"group-main-agent:{group_id}"
        if owner:
            agent_id = f"owner-agent:{user_id}"
            layers.append(ContextLayer("owner_policy", 90, owner_policy or {}))
        resolved: dict[str, Any] = {}
        # Higher priority overlays lower priority except safety policies handled upstream.
        for layer in sorted(layers, key=lambda x: x.priority):
            resolved.update(layer.data)
        return EffectiveContext(agent_id, user_id, group_id, layers, resolved)
 __all__ = ["ContextLayer", "EffectiveContext", "AgentContextResolver"]
@@ -1,77 +0,0 @@
 """Actions API for unified Agent Event Bus."""
 from __future__ import annotations
 from typing import Any
 try:
    from fastapi import APIRouter
 except Exception:
    APIRouter = None
 from .agent_event_bus import AgentEventBus
 bus = AgentEventBus()
 def create_agent_event(payload: dict[str, Any]) -> dict[str, Any]:
    return {"status": "ok", "event": bus.publish(**payload)}
 def list_agent_events(kind: str | None = None, status: str = "pending") -> dict[str, Any]:
    return {"status": "ok", "events": bus.list_events(kind, status)}
 def get_agent_event(event_id: str) -> dict[str, Any]:
    return {"status": "ok", "event": bus.get(event_id)}
 def transition_agent_event(event_id: str, status: str, actor: str = "system") -> dict[str, Any]:
    return {"status": "ok", "event": bus.transition(event_id, status, actor)}
 def agent_event_stats() -> dict[str, Any]:
    return {"status": "ok", "stats": bus.stats()}
 # Compatibility adapters
 def list_gpt_inbox_messages():
    return list_agent_events(kind="gpt_inbox")
 def ack_gpt_inbox_message(event_id: str):
    return transition_agent_event(event_id, "completed", "custom-gpt")
 def build_router():
    if APIRouter is None:
        return None
    router = APIRouter(prefix="/agent-events", tags=["agent-events"])
    @router.post("", operation_id="createAgentEvent")
    def create(payload: dict[str, Any]):
        return create_agent_event(payload)
    @router.get("", operation_id="listAgentEvents")
    def listing(kind: str | None = None, status: str = "pending"):
        return list_agent_events(kind, status)
    @router.get("/{event_id}", operation_id="getAgentEvent")
    def get(event_id: str):
        return get_agent_event(event_id)
    @router.post("/{event_id}/transition", operation_id="transitionAgentEvent")
    def transition(event_id: str, payload: dict[str, Any]):
        return transition_agent_event(event_id, payload["status"], payload.get("actor", "system"))
    @router.get("/stats/overview", operation_id="getAgentEventStats")
    def stats():
        return agent_event_stats()
    return router
 router = build_router()
 __all__ = ["router", "create_agent_event", "list_agent_events", "get_agent_event", "transition_agent_event"]
@@ -1,125 +0,0 @@
 """Unified Agent Event Bus.
 Unifies commands, GPT messages, replies, approvals, audit and reports into one
 state machine. Compatibility layers can filter events by kind instead of keeping
 separate queues.
 """
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from hashlib import sha256
 from pathlib import Path
 import json
 from typing import Any
 def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()
@dataclass
 class AgentEvent:
    id: str
    kind: str
    source: str
    target: str
    summary: str
    payload: dict[str, Any]
    status: str = "pending"
    correlation_id: str | None = None
    created_at: str = field(default_factory=utc_now)
    updated_at: str = field(default_factory=utc_now)
    actor: str | None = None
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
 class AgentEventBus:
    STATES = ("pending", "running", "completed", "failed")
    def __init__(self, root: str = "agents/events"):
        self.root = Path(root)
        for state in self.STATES:
            (self.root / state).mkdir(parents=True, exist_ok=True)
        (self.root / "audit").mkdir(parents=True, exist_ok=True)
    def publish(
        self,
        kind: str,
        source: str,
        target: str,
        summary: str,
        payload: dict[str, Any],
        correlation_id: str | None = None,
    ) -> dict[str, Any]:
        event_id = "evt_" + sha256(
            json.dumps(
                {"kind": kind, "source": source, "summary": summary, "payload": payload},
                sort_keys=True,
                ensure_ascii=False,
            ).encode("utf-8")
        ).hexdigest()[:16]
        event = AgentEvent(event_id, kind, source, target, summary, payload, correlation_id=correlation_id)
        self._write("pending", event.to_dict())
        self._audit("created", event.to_dict())
        return event.to_dict()
    def list_events(self, kind: str | None = None, status: str = "pending", limit: int = 100) -> list[dict[str, Any]]:
        result = []
        for path in sorted((self.root / status).glob("*.json")):
            event = self._read(path)
            if kind and event.get("kind") != kind:
                continue
            result.append(event)
        return result[:limit]
    def get(self, event_id: str) -> dict[str, Any] | None:
        for state in self.STATES:
            path = self.root / state / f"{event_id}.json"
            if path.exists():
                return self._read(path)
        return None
    def transition(self, event_id: str, status: str, actor: str | None = None) -> dict[str, Any]:
        event = self.get(event_id)
        if not event:
            return {"status": "not_found", "id": event_id}
        if status not in self.STATES:
            return {"status": "invalid_state"}
        old = event.get("status", "pending")
        old_path = self.root / old / f"{event_id}.json"
        event["status"] = status
        event["actor"] = actor
        event["updated_at"] = utc_now()
        self._write(status, event)
        if old_path.exists() and old != status:
            old_path.unlink()
        self._audit("transition", event, {"from": old, "to": status})
        return event
    def stats(self) -> dict[str, Any]:
        return {
            "events": {state: len(list((self.root / state).glob("*.json"))) for state in self.STATES},
            "audit_events": len(list((self.root / "audit").glob("*.json"))),
            "status": "healthy",
        }
    def _write(self, state: str, data: dict[str, Any]):
        (self.root / state / f"{data['id']}.json").write_text(
            json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8"
        )
    def _read(self, path: Path):
        return json.loads(path.read_text(encoding="utf-8"))
    def _audit(self, action: str, event: dict[str, Any], extra: dict[str, Any] | None = None):
        data = {"action": action, "event": event, "extra": extra or {}, "time": utc_now()}
        (self.root / "audit" / f"{event['id']}-{action}.json").write_text(
            json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8"
        )
 __all__ = ["AgentEvent", "AgentEventBus"]
@@ -1,50 +0,0 @@
 """Admin dashboard data model for the unified Agent Event Bus."""
 from __future__ import annotations
 from collections import Counter
 from typing import Any
 from .agent_event_bus import AgentEventBus
 class AgentEventDashboard:
    def __init__(self, bus: AgentEventBus | None = None):
        self.bus = bus or AgentEventBus()
    def overview(self) -> dict[str, Any]:
        stats = self.bus.stats()
        pending = self.bus.list_events(status="pending")
        running = self.bus.list_events(status="running")
        completed = self.bus.list_events(status="completed")
        failed = self.bus.list_events(status="failed")
        all_visible = pending + running + completed + failed
        by_kind = Counter(event.get("kind", "unknown") for event in all_visible)
        by_target = Counter(event.get("target", "unknown") for event in all_visible)
        return {
            "module": "agent-event-bus",
            "title": "Unified Agent Event Bus",
            "route": "/admin/ai-evolution/events",
            "metrics": {
                **stats,
                "by_kind": dict(by_kind),
                "by_target": dict(by_target),
                "gpt_inbox_pending": by_kind.get("gpt_inbox", 0),
                "openclaw_pending": len([e for e in pending if e.get("target") == "openclaw"]),
            },
            "panels": [
                "event timeline",
                "pending commands",
                "GPT inbox",
                "agent replies",
                "failure reports",
                "audit trail",
            ],
            "audit": {
                "enabled": True,
                "location": "agents/events/audit",
            },
        }
 __all__ = ["AgentEventDashboard"]
@@ -1,37 +0,0 @@
 """Bridge Agent Event Bus events into Agent Observability metrics."""
 from .agent_observability_hooks import AgentObservabilityHooks
 class AgentEventObservabilityBridge:
    def __init__(self, hooks: AgentObservabilityHooks | None = None):
        self.hooks = hooks or AgentObservabilityHooks()
    def consume(self, event: dict):
        kind = event.get("kind")
        payload = event.get("payload", {})
        if kind == "agent_started":
            return self.hooks.start(
                user_id=payload.get("user_id", "unknown"),
                visible_scope=payload.get("scope", "self"),
                entry_agent=payload.get("agent", "unknown"),
                execution_id=event.get("id"),
            )
        if kind == "agent_delegated":
            return self.hooks.delegate(event["execution_id"], payload.get("agent", "unknown"))
        if kind == "token_consumed":
            return self.hooks.tokens(
                event["execution_id"],
                payload.get("input_tokens", 0),
                payload.get("output_tokens", 0),
            )
        if kind in {"agent_completed", "agent_failed"}:
            return self.hooks.finish(
                event["execution_id"],
                payload.get("closed_by"),
                "failed" if kind == "agent_failed" else "completed",
            )
        return {"status": "ignored", "kind": kind}
 __all__ = ["AgentEventObservabilityBridge"]
@@ -0,0 +1,75 @@
 """Actions API layer for Agent GPT message plane."""
 from __future__ import annotations
 from typing import Any
 try:
    from fastapi import APIRouter
 except Exception:
    APIRouter = None
 from .agent_gpt_message_plane import AgentGPTMessagePlane
 plane = AgentGPTMessagePlane()
 def list_gpt_inbox(limit: int = 50) -> dict[str, Any]:
    return {"status": "ok", "messages": plane.list_inbox(limit)}
 def get_gpt_message(message_id: str) -> dict[str, Any]:
    msg = plane.get_message(message_id)
    return {"status": "ok", "message": msg} if msg else {"status": "not_found"}
 def ack_gpt_message(message_id: str) -> dict[str, Any]:
    return plane.acknowledge(message_id)
 def reply_to_agent(payload: dict[str, Any]) -> dict[str, Any]:
    return {"status": "ok", "message": plane.reply_from_gpt(payload)}
 def message_plane_statistics() -> dict[str, Any]:
    return {"status": "ok", "statistics": plane.statistics()}
 def build_router():
    if APIRouter is None:
        return None
    router = APIRouter(prefix="/agent-gpt", tags=["agent-gpt"])
    @router.get("/inbox", operation_id="listGptInboxMessages")
    def inbox():
        return list_gpt_inbox()
    @router.get("/messages/{message_id}", operation_id="getGptInboxMessage")
    def get_message(message_id: str):
        return get_gpt_message(message_id)
    @router.post("/messages/{message_id}/ack", operation_id="ackGptInboxMessage")
    def ack(message_id: str):
        return ack_gpt_message(message_id)
    @router.post("/reply", operation_id="replyToAgentMessage")
    def reply(payload: dict[str, Any]):
        return reply_to_agent(payload)
    @router.get("/stats", operation_id="getAgentGptMessageStats")
    def stats():
        return message_plane_statistics()
    return router
 router = build_router()
 __all__ = [
    "router",
    "list_gpt_inbox",
    "get_gpt_message",
    "ack_gpt_message",
    "reply_to_agent",
    "message_plane_statistics",
 ]
@@ -0,0 +1,29 @@
 """Admin Web UI data model for Agent GPT message plane."""
 from .agent_gpt_message_plane import AgentGPTMessagePlane
 class AgentGPTMessageDashboard:
    def __init__(self, plane: AgentGPTMessagePlane | None = None):
        self.plane = plane or AgentGPTMessagePlane()
    def overview(self):
        stats = self.plane.statistics()
        return {
            "module": "agent-gpt-message-plane",
            "title": "Agent ↔ Custom GPT Message Plane",
            "metrics": stats,
            "audit": {
                "enabled": True,
                "location": "agents/messages/audit",
            },
            "panels": [
                "message inbox timeline",
                "OpenClaw executions",
                "GPT acknowledgements",
                "failed message flows",
            ],
        }
 __all__ = ["AgentGPTMessageDashboard"]
@@ -0,0 +1,144 @@
 """Bidirectional OpenClaw <-> Custom GPT message plane.
 Provides an auditable inbox/outbox model. It does not push into a ChatGPT
 conversation directly; Custom GPT reads messages through Actions APIs.
 """
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from hashlib import sha256
 from pathlib import Path
 import json
 from typing import Any
 def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()
 def message_id(payload: dict[str, Any]) -> str:
    raw = json.dumps(payload, sort_keys=True, ensure_ascii=False)
    return "msg_" + sha256(raw.encode("utf-8")).hexdigest()[:16]
@dataclass
 class AgentMessage:
    id: str
    source: str
    target: str
    message_type: str
    summary: str
    payload: dict[str, Any]
    status: str = "pending"
    correlation_id: str | None = None
    created_at: str = field(default_factory=utc_now)
    updated_at: str = field(default_factory=utc_now)
    ack_by: str | None = None
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
 class AgentGPTMessagePlane:
    def __init__(self, root: str = "agents/messages"):
        self.root = Path(root)
        self.pending = self.root / "to-gpt" / "pending"
        self.completed = self.root / "to-gpt" / "completed"
        self.reply_pending = self.root / "from-gpt" / "pending"
        self.audit = self.root / "audit"
        for path in [self.pending, self.completed, self.reply_pending, self.audit]:
            path.mkdir(parents=True, exist_ok=True)
    def send_to_gpt(
        self,
        source: str,
        summary: str,
        payload: dict[str, Any],
        correlation_id: str | None = None,
        message_type: str = "status_report",
    ) -> dict[str, Any]:
        msg_id = message_id({"source": source, "summary": summary, "payload": payload})
        msg = AgentMessage(
            id=msg_id,
            source=source,
            target="custom-gpt",
            message_type=message_type,
            summary=summary,
            payload=payload,
            correlation_id=correlation_id,
        )
        self._write(self.pending, msg)
        self._audit("message_created", msg)
        return msg.to_dict()
    def list_inbox(self, limit: int = 50) -> list[dict[str, Any]]:
        messages = [self._read(path) for path in sorted(self.pending.glob("*.json"))]
        return [m for m in messages[:limit]]
    def get_message(self, msg_id: str) -> dict[str, Any] | None:
        for path in self.pending.glob(f"{msg_id}.json"):
            return self._read(path)
        for path in self.completed.glob(f"{msg_id}.json"):
            return self._read(path)
        return None
    def acknowledge(self, msg_id: str, actor: str = "custom-gpt") -> dict[str, Any]:
        msg = self.get_message(msg_id)
        if not msg:
            return {"status": "not_found", "id": msg_id}
        msg["status"] = "completed"
        msg["ack_by"] = actor
        msg["updated_at"] = utc_now()
        self._write_json(self.completed / f"{msg_id}.json", msg)
        pending_file = self.pending / f"{msg_id}.json"
        if pending_file.exists():
            pending_file.unlink()
        self._audit("message_ack", msg)
        return {"status": "acked", "message": msg}
    def reply_from_gpt(self, payload: dict[str, Any]) -> dict[str, Any]:
        msg_id = message_id(payload)
        data = {
            "id": msg_id,
            "source": "custom-gpt",
            "target": payload.get("target", "openclaw"),
            "status": "pending",
            "payload": payload,
            "created_at": utc_now(),
        }
        self._write_json(self.reply_pending / f"{msg_id}.json", data)
        self._write_json(self.audit / f"{msg_id}.json", {"event": "reply_created", **data})
        return data
    def statistics(self) -> dict[str, Any]:
        pending = len(list(self.pending.glob("*.json")))
        completed = len(list(self.completed.glob("*.json")))
        replies = len(list(self.reply_pending.glob("*.json")))
        audits = len(list(self.audit.glob("*.json")))
        return {
            "pending_messages": pending,
            "completed_messages": completed,
            "pending_replies": replies,
            "audit_events": audits,
            "status": "healthy",
        }
    def _write(self, folder: Path, msg: AgentMessage):
        self._write_json(folder / f"{msg.id}.json", msg.to_dict())
    def _write_json(self, path: Path, data: dict[str, Any]):
        path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
    def _read(self, path: Path) -> dict[str, Any]:
        return json.loads(path.read_text(encoding="utf-8"))
    def _audit(self, event: str, msg: AgentMessage):
        self._write_json(
            self.audit / f"{msg.id}-{event}.json",
            {"event": event, "message": msg.to_dict()},
        )
 __all__ = ["AgentMessage", "AgentGPTMessagePlane"]
@@ -1,80 +0,0 @@
 """Agent orchestration observability model.
 Tracks OpenClaw, Hermes and sub-agent execution chains. Token usage is optional
 and depends on model adapter instrumentation.
 """
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from typing import Any
 def utc_now():
    return datetime.now(timezone.utc).isoformat()
@dataclass
 class AgentExecutionRecord:
    execution_id: str
    user_id: str
    visible_scope: str
    entry_agent: str
    path: list[str]
    used_hermes: bool
    closed_by: str
    duration_ms: int = 0
    input_tokens: int | None = None
    output_tokens: int | None = None
    total_tokens: int | None = None
    status: str = "completed"
    started_at: str = field(default_factory=utc_now)
    ended_at: str = field(default_factory=utc_now)
    def to_dict(self):
        return asdict(self)
 class AgentObservability:
    def __init__(self):
        self.records: list[AgentExecutionRecord] = []
    def record(self, record: AgentExecutionRecord):
        self.records.append(record)
        return record.to_dict()
    def summary(self, user_id: str | None = None) -> dict[str, Any]:
        records = [r for r in self.records if not user_id or r.user_id == user_id]
        hermes = [r for r in records if r.used_hermes]
        direct = [r for r in records if not r.used_hermes]
        return {
            "total_executions": len(records),
            "openclaw_independent_loops": len(direct),
            "hermes_orchestrated_loops": len(hermes),
            "average_duration_ms": (
                sum(r.duration_ms for r in records) / len(records) if records else 0
            ),
            "token_usage": {
                "input": sum(r.input_tokens or 0 for r in records),
                "output": sum(r.output_tokens or 0 for r in records),
                "total": sum(r.total_tokens or 0 for r in records),
            },
        }
    def hermes_tree(self, user_id: str | None = None):
        return [
            {
                "execution_id": r.execution_id,
                "entry": r.entry_agent,
                "used_hermes": r.used_hermes,
                "children": r.path,
                "duration_ms": r.duration_ms,
                "tokens": r.total_tokens,
            }
            for r in self.records
            if (not user_id or r.user_id == user_id)
        ]
 __all__ = ["AgentExecutionRecord", "AgentObservability"]
@@ -1,52 +0,0 @@
 """API for agent orchestration observability."""
 from __future__ import annotations
 from typing import Any
 try:
    from fastapi import APIRouter
 except Exception:
    APIRouter = None
 from .agent_observability import AgentExecutionRecord
 from .agent_observability_dashboard import AgentObservabilityDashboard
 from .agent_observability_store import AgentObservabilityStore
 store = AgentObservabilityStore()
 dashboard = AgentObservabilityDashboard(store.load())
 def record_agent_execution(payload: dict[str, Any]):
    record = AgentExecutionRecord(**payload)
    return {"status": "ok", "record": store.append(record)}
 def get_agent_observability(user_id: str | None = None, group_id: str | None = None, role: str = "user"):
    records = store.query(user_id, group_id, role)
    return {
        "status": "ok",
        "records": records,
        "summary": store.load().summary(user_id if role != "owner" else None),
    }
 def build_router():
    if APIRouter is None:
        return None
    router = APIRouter(prefix="/admin/ai-evolution/agents", tags=["agent-observability"])
    @router.post("/execution", operation_id="recordAgentExecution")
    def record(payload: dict[str, Any]):
        return record_agent_execution(payload)
    @router.get("/overview", operation_id="getAgentObservabilityOverview")
    def overview(user_id: str | None = None, group_id: str | None = None, role: str = "user"):
        return get_agent_observability(user_id, group_id, role)
    return router
 router = build_router()
 __all__ = ["router", "record_agent_execution", "get_agent_observability"]
@@ -1,44 +0,0 @@
 """Web UI dashboard model for agent execution visibility."""
 from .agent_observability import AgentObservability
 class AgentObservabilityDashboard:
    def __init__(self, observability: AgentObservability | None = None):
        self.observability = observability or AgentObservability()
    def overview(self, user_id: str | None = None):
        return {
            "route": "/admin/ai-evolution/agents",
            "title": "Agent Orchestration Observability",
            "summary": self.observability.summary(user_id),
            "views": [
                {
                    "name": "scheduler_flow",
                    "title": "OpenClaw/Hermes 调度流程",
                    "data": self.observability.hermes_tree(user_id),
                },
                {
                    "name": "agent_metrics",
                    "title": "智能体指标",
                    "fields": [
                        "duration_ms",
                        "input_tokens",
                        "output_tokens",
                        "total_tokens",
                        "status",
                    ],
                },
                {
                    "name": "loop_distribution",
                    "title": "闭环分布",
                    "fields": [
                        "openclaw_independent_loops",
                        "hermes_orchestrated_loops",
                    ],
                },
            ],
        }
 __all__ = ["AgentObservabilityDashboard"]
@@ -1,113 +0,0 @@
 """Runtime hooks for Agent Observability.
 OpenClaw/Hermes/sub-agents can call these hooks at start/end/delegation points.
 The hooks store sanitized metadata only: no prompts, no secrets, no private message
 content.
 """
 from __future__ import annotations
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
 from typing import Any
 from uuid import uuid4
 from .agent_observability import AgentExecutionRecord
 from .agent_observability_store import AgentObservabilityStore
 def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()
 def parse_time(value: str) -> datetime:
    return datetime.fromisoformat(value.replace("Z", "+00:00"))
@dataclass
 class AgentRunContext:
    execution_id: str
    user_id: str
    visible_scope: str
    entry_agent: str
    path: list[str] = field(default_factory=list)
    used_hermes: bool = False
    started_at: str = field(default_factory=utc_now)
    input_tokens: int = 0
    output_tokens: int = 0
    status: str = "running"
    def delegate(self, agent_name: str) -> None:
        if agent_name not in self.path:
            self.path.append(agent_name)
        if agent_name.lower() == "hermes":
            self.used_hermes = True
    def add_tokens(self, input_tokens: int = 0, output_tokens: int = 0) -> None:
        self.input_tokens += max(input_tokens, 0)
        self.output_tokens += max(output_tokens, 0)
 class AgentObservabilityHooks:
    def __init__(self, store: AgentObservabilityStore | None = None):
        self.store = store or AgentObservabilityStore()
        self.active: dict[str, AgentRunContext] = {}
    def start(
        self,
        user_id: str,
        visible_scope: str,
        entry_agent: str,
        execution_id: str | None = None,
    ) -> dict[str, Any]:
        run = AgentRunContext(
            execution_id=execution_id or f"exec_{uuid4().hex[:12]}",
            user_id=user_id,
            visible_scope=visible_scope,
            entry_agent=entry_agent,
            path=[entry_agent],
            used_hermes=entry_agent.lower() == "hermes",
        )
        self.active[run.execution_id] = run
        return {"status": "started", "execution_id": run.execution_id, "run": run.__dict__}
    def delegate(self, execution_id: str, agent_name: str) -> dict[str, Any]:
        run = self.active[execution_id]
        run.delegate(agent_name)
        return {"status": "delegated", "execution_id": execution_id, "path": run.path}
    def tokens(self, execution_id: str, input_tokens: int = 0, output_tokens: int = 0) -> dict[str, Any]:
        run = self.active[execution_id]
        run.add_tokens(input_tokens, output_tokens)
        return {
            "status": "tokens_recorded",
            "execution_id": execution_id,
            "input_tokens": run.input_tokens,
            "output_tokens": run.output_tokens,
            "total_tokens": run.input_tokens + run.output_tokens,
        }
    def finish(self, execution_id: str, closed_by: str | None = None, status: str = "completed") -> dict[str, Any]:
        run = self.active.pop(execution_id)
        ended_at = utc_now()
        duration_ms = int((parse_time(ended_at) - parse_time(run.started_at)).total_seconds() * 1000)
        record = AgentExecutionRecord(
            execution_id=run.execution_id,
            user_id=run.user_id,
            visible_scope=run.visible_scope,
            entry_agent=run.entry_agent,
            path=run.path,
            used_hermes=run.used_hermes,
            closed_by=closed_by or run.path[-1],
            duration_ms=duration_ms,
            input_tokens=run.input_tokens,
            output_tokens=run.output_tokens,
            total_tokens=run.input_tokens + run.output_tokens,
            status=status,
            started_at=run.started_at,
            ended_at=ended_at,
        )
        return {"status": "stored", "record": self.store.append(record)}
 __all__ = ["AgentRunContext", "AgentObservabilityHooks"]
@@ -1,45 +0,0 @@
 """Persistent store for agent observability records.
 Stores sanitized execution metrics for OpenClaw/Hermes/sub-agents. No prompts,
 secrets, or private message contents are persisted here; only metrics and routing
 metadata are stored for Web UI statistics and audit.
 """
 from __future__ import annotations
 import json
 from pathlib import Path
 from typing import Any
 from .agent_observability import AgentExecutionRecord, AgentObservability
 class AgentObservabilityStore:
    def __init__(self, root: str = "runtime/agent-observability"):
        self.root = Path(root)
        self.records_dir = self.root / "records"
        self.records_dir.mkdir(parents=True, exist_ok=True)
    def append(self, record: AgentExecutionRecord) -> dict[str, Any]:
        path = self.records_dir / f"{record.execution_id}.json"
        data = record.to_dict()
        path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
        return data
    def load(self) -> AgentObservability:
        observability = AgentObservability()
        for path in sorted(self.records_dir.glob("*.json")):
            raw = json.loads(path.read_text(encoding="utf-8"))
            observability.records.append(AgentExecutionRecord(**raw))
        return observability
    def query(self, user_id: str | None = None, group_id: str | None = None, role: str = "user") -> list[dict[str, Any]]:
        records = [record.to_dict() for record in self.load().records]
        if role == "owner":
            return records
        if role == "group_admin" and group_id:
            return [r for r in records if r.get("visible_scope") == group_id or r.get("user_id") == user_id]
        return [r for r in records if r.get("user_id") == user_id]
 __all__ = ["AgentObservabilityStore"]
@@ -15,7 +15,7 @@ TOKEN_FILE = Path(os.getenv('AIOS_GITOPS_TOKEN_FILE', str(SECRETS / 'gitops.toke
 HOST = os.getenv('AIOS_GITOPS_HOST', '127.0.0.1')
 PORT = int(os.getenv('AIOS_GITOPS_PORT', '18890'))
 OPENAPI = WS / 'docs' / 'gitops' / 'openapi-chatgpt-actions.yaml'
-BARE_REPO = Path(os.getenv('AIOS_GITOPS_BARE_REPO', str(ROOT / 'runtime' / 'gitea-bin' / 'data' / 'repositories' / 'aios' / 'aios.git')))
+BARE_REPO = Path(os.getenv('AIOS_GITOPS_BARE_REPO', str(ROOT / 'runtime' / 'gitea-bin' / 'data' / 'repositories' / 'aios' / 'aios-v23.git')))
 monitor = ActionMonitor()
 BASE_MODELS = [
@@ -1,89 +0,0 @@
 """Append-only audit chain for AIOS V2R1."""
 from __future__ import annotations
 import hashlib
 import json
 from pathlib import Path
 import sqlite3
 from typing import Any
 from .tenant_context import canonical_json, isoformat, require_ident
 class AuditLog:
    def __init__(self, database: str | Path) -> None:
        self.database = str(database)
        with sqlite3.connect(self.database) as db:
            db.execute("PRAGMA journal_mode=WAL")
            db.execute(
                """
                CREATE TABLE IF NOT EXISTS audit_events (
                    seq INTEGER PRIMARY KEY AUTOINCREMENT,
                    tenant_id TEXT NOT NULL,
                    trace_id TEXT NOT NULL,
                    event_type TEXT NOT NULL,
                    payload TEXT NOT NULL,
                    previous_hash TEXT NOT NULL,
                    event_hash TEXT NOT NULL UNIQUE,
                    created_at TEXT NOT NULL
                )
                """
            )
    def append(self, *, tenant_id: str, trace_id: str, event_type: str, payload: dict[str, Any]) -> str:
        tenant_id = require_ident("tenant_id", tenant_id, tenant=True)
        trace_id = require_ident("trace_id", trace_id)
        event_type = require_ident("event_type", event_type)
        with sqlite3.connect(self.database) as db:
            previous = db.execute(
                "SELECT event_hash FROM audit_events ORDER BY seq DESC LIMIT 1"
            ).fetchone()
            previous_hash = previous[0] if previous else "GENESIS"
            body = {
                "tenant_id": tenant_id,
                "trace_id": trace_id,
                "event_type": event_type,
                "payload": payload,
                "previous_hash": previous_hash,
                "created_at": isoformat(),
            }
            digest = hashlib.sha256(
                canonical_json(body)
            ).hexdigest()
            db.execute(
                "INSERT INTO audit_events VALUES(NULL,?,?,?,?,?,?,?)",
                (
                    tenant_id,
                    trace_id,
                    event_type,
                    json.dumps(payload, ensure_ascii=False),
                    previous_hash,
                    digest,
                    body["created_at"],
                ),
            )
            return digest
    def verify(self) -> bool:
        previous = "GENESIS"
        with sqlite3.connect(self.database) as db:
            rows = db.execute(
                "SELECT tenant_id,trace_id,event_type,payload,previous_hash,event_hash,created_at "
                "FROM audit_events ORDER BY seq"
            )
            for row in rows:
                body = {
                    "tenant_id": row[0],
                    "trace_id": row[1],
                    "event_type": row[2],
                    "payload": json.loads(row[3]),
                    "previous_hash": row[4],
                    "created_at": row[6],
                }
                if row[4] != previous:
                    return False
                digest = hashlib.sha256(canonical_json(body)).hexdigest()
                if digest != row[5]:
                    return False
                previous = digest
        return True
@@ -1,73 +0,0 @@
 """AIOS V2R1 Tenant Cell lifecycle CLI primitives."""
 from __future__ import annotations
 from dataclasses import dataclass
 from datetime import datetime, timezone
 import json
 from pathlib import Path
 from .tenant_context import ValidationError, require_ident
@dataclass(frozen=True)
 class CellRecord:
    tenant_id: str
    cell_id: str
    version: str
    state: str
    config_digest: str
    created_at: str
 class CellManager:
    def __init__(self, registry: str | Path) -> None:
        self.registry = Path(registry)
        self.registry.parent.mkdir(parents=True, exist_ok=True)
        if not self.registry.exists():
            self.registry.write_text("{\"cells\":{}}\n", encoding="utf-8")
    def _load(self) -> dict:
        return json.loads(self.registry.read_text(encoding="utf-8"))
    def _save(self, value: dict) -> None:
        self.registry.write_text(
            json.dumps(value, ensure_ascii=False, indent=2) + "\n",
            encoding="utf-8",
        )
    def create(self, tenant_id: str, cell_id: str, version: str, digest: str) -> CellRecord:
        tenant_id = require_ident("tenant_id", tenant_id, tenant=True)
        cell_id = require_ident("cell_id", cell_id)
        data = self._load()
        if tenant_id in data["cells"]:
            raise ValidationError("tenant already has a cell")
        record = CellRecord(
            tenant_id,
            cell_id,
            version,
            "created",
            digest,
            datetime.now(timezone.utc).isoformat(),
        )
        data["cells"][tenant_id] = record.__dict__
        self._save(data)
        return record
    def status(self, tenant_id: str) -> CellRecord:
        data = self._load()
        item = data["cells"].get(tenant_id)
        if not item:
            raise ValidationError("unknown tenant")
        return CellRecord(**item)
    def transition(self, tenant_id: str, state: str) -> CellRecord:
        allowed = {"created", "starting", "healthy", "upgrading", "rollback", "failed"}
        if state not in allowed:
            raise ValidationError("invalid cell state")
        data = self._load()
        item = data["cells"].get(tenant_id)
        if not item:
            raise ValidationError("unknown tenant")
        item["state"] = state
        self._save(data)
        return CellRecord(**item)
@@ -1,64 +0,0 @@
 """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)
@@ -1,48 +0,0 @@
 """aios-cell command implementation for V2R1 Cell lifecycle."""
 from __future__ import annotations
 import argparse
 import json
 import sys
 from .cell_manager import CellManager
 def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(prog="aios-cell")
    parser.add_argument("registry")
    sub = parser.add_subparsers(dest="command", required=True)
    create = sub.add_parser("create")
    create.add_argument("tenant_id")
    create.add_argument("cell_id")
    create.add_argument("version")
    create.add_argument("digest")
    status = sub.add_parser("status")
    status.add_argument("tenant_id")
    transition = sub.add_parser("transition")
    transition.add_argument("tenant_id")
    transition.add_argument("state")
    return parser
 def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    manager = CellManager(args.registry)
    if args.command == "create":
        result = manager.create(args.tenant_id, args.cell_id, args.version, args.digest)
    elif args.command == "status":
        result = manager.status(args.tenant_id)
    elif args.command == "transition":
        result = manager.transition(args.tenant_id, args.state)
    else:
        return 2
    print(json.dumps(result.__dict__, ensure_ascii=False, sort_keys=True))
    return 0
 if __name__ == "__main__":
    sys.exit(main())
@@ -1,106 +0,0 @@
 """OpenClaw Cell lifecycle adapter for AIOS V2R1."""
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 import json
 from pathlib import Path
 import subprocess
 from typing import Any, Mapping
 from .tenant_context import ValidationError, V2R1Error, require_ident
@dataclass(frozen=True)
 class CellSpec:
    tenant_id: str
    image: str
    runtime: str = "docker"
    port: int | None = None
    memory: str = "2g"
    cpus: str = "2"
    pids_limit: int = 512
    network: str = "bridge"
    env: tuple[str, ...] = field(default_factory=tuple)
    def __post_init__(self) -> None:
        require_ident("tenant_id", self.tenant_id, tenant=True)
        if self.runtime not in {"docker", "podman"}:
            raise ValidationError("unsupported runtime")
        if not self.image:
            raise ValidationError("image required")
 class OpenClawFleet:
    def __init__(self, binary: str = "openclaw") -> None:
        self.binary = binary
    def _run(self, args: list[str]) -> dict[str, Any]:
        proc = subprocess.run(
            [self.binary, *args],
            check=False,
            text=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )
        if proc.returncode:
            raise V2R1Error(proc.stderr.strip() or "openclaw command failed")
        data = json.loads(proc.stdout or "{}")
        if isinstance(data, dict):
            return data
        return {"result": data}
    def create(self, spec: CellSpec) -> dict[str, Any]:
        return self._run([
            "fleet", "create", spec.tenant_id,
            "--image", spec.image,
            "--runtime", spec.runtime,
            "--json",
        ])
    def status(self, tenant_id: str) -> dict[str, Any]:
        require_ident("tenant_id", tenant_id, tenant=True)
        return self._run(["fleet", "status", tenant_id, "--json"])
    def doctor(self, tenant_id: str | None = None) -> dict[str, Any]:
        args = ["fleet", "doctor"]
        if tenant_id:
            require_ident("tenant_id", tenant_id, tenant=True)
            args.append(tenant_id)
        args.append("--json")
        return self._run(args)
    def backup(self, tenant_id: str, output: str | Path) -> dict[str, Any]:
        return self._run(["fleet", "backup", tenant_id, "--out", str(output), "--json"])
    def restore(self, tenant_id: str, archive: str | Path) -> dict[str, Any]:
        return self._run(["fleet", "restore", tenant_id, "--from", str(archive), "--json"])
    def upgrade(self, tenant_id: str, image: str) -> dict[str, Any]:
        return self._run(["fleet", "upgrade", tenant_id, "--image", image, "--json"])
 def load_inventory(path: str | Path) -> list[CellSpec]:
    raw = json.loads(Path(path).read_text(encoding="utf-8"))
    if raw.get("schema_version") != "aios.cells.v1":
        raise ValidationError("invalid inventory schema")
    result: list[CellSpec] = []
    seen: set[str] = set()
    for item in raw.get("cells", []):
        spec = CellSpec(
            tenant_id=item["tenant_id"],
            image=item["image"],
            runtime=item.get("runtime", "docker"),
            port=item.get("port"),
            env=tuple(item.get("env", [])),
        )
        if spec.tenant_id in seen:
            raise ValidationError("duplicate tenant")
        seen.add(spec.tenant_id)
        result.append(spec)
    if not result:
        raise ValidationError("no cells")
    return result
 def inventory_as_dict(items: list[CellSpec]) -> dict[str, Any]:
    return {"schema_version": "aios.cells.v1", "cells": [asdict(x) for x in items]}
@@ -1,143 +0,0 @@
 """Tenant-partitioned Hermes shared agent fabric primitives."""
 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
 from .tenant_context import (
    CanonicalMessage,
    ConflictError,
    ScopeError,
    ValidationError,
    isoformat,
    require_ident,
 )
@dataclass(frozen=True)
 class FabricJob:
    job_id: str
    tenant_id: str
    cell_id: str
    profile_id: str
    session_id: str
    event_id: str
    status: str = "queued"
    def __post_init__(self) -> None:
        for name in (
            "job_id", "cell_id", "profile_id", "session_id", "event_id"
        ):
            require_ident(name, getattr(self, name))
        require_ident("tenant_id", self.tenant_id, tenant=True)
        if self.status not in {"queued", "running", "completed", "failed", "cancelled"}:
            raise ValidationError("invalid job status")
 class HermesFabric:
    """Shared scheduler storage with mandatory tenant partition keys."""
    def __init__(self, database: str | Path) -> None:
        self.database = str(database)
        with closing(sqlite3.connect(self.database)) as connection, connection as db:
            db.execute("PRAGMA journal_mode=WAL")
            db.execute("PRAGMA synchronous=FULL")
            db.execute(
                """
                CREATE TABLE IF NOT EXISTS fabric_jobs (
                    job_id TEXT PRIMARY KEY,
                    tenant_id TEXT NOT NULL,
                    cell_id TEXT NOT NULL,
                    profile_id TEXT NOT NULL,
                    session_id TEXT NOT NULL,
                    event_id TEXT NOT NULL UNIQUE,
                    status TEXT NOT NULL,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL
                )
                """
            )
            db.execute(
                "CREATE INDEX IF NOT EXISTS idx_fabric_tenant_queue "
                "ON fabric_jobs(tenant_id, cell_id, status)"
            )
    def enqueue(
        self,
        *,
        message: CanonicalMessage,
        profile_id: str,
    ) -> FabricJob:
        profile_id = require_ident("profile_id", profile_id)
        if not message.command_capable:
            raise ScopeError("non-command events cannot create execution jobs")
        if not message.is_authenticated_human_command:
            raise ScopeError("only authenticated human commands can create jobs")
        job = FabricJob(
            job_id=f"job-{message.event_id}",
            tenant_id=message.tenant_id,
            cell_id=message.cell_id,
            profile_id=profile_id,
            session_id=message.session_id,
            event_id=message.event_id,
        )
        now = isoformat()
        conflict: ConflictError | None = None
        try:
            with closing(sqlite3.connect(self.database)) as connection, connection as db:
                db.execute("PRAGMA journal_mode=WAL")
                db.execute(
                    "INSERT INTO fabric_jobs VALUES(?,?,?,?,?,?,?,?,?)",
                    (
                        job.job_id,
                        job.tenant_id,
                        job.cell_id,
                        job.profile_id,
                        job.session_id,
                        job.event_id,
                        job.status,
                        now,
                        now,
                    ),
                )
        except sqlite3.IntegrityError:
            conflict = ConflictError(
                "fabric event already exists or violates queue uniqueness"
            )
        if conflict is not None:
            raise conflict
        return job
    def claim(self, tenant_id: str, cell_id: str, profile_id: str) -> FabricJob | None:
        require_ident("tenant_id", tenant_id, tenant=True)
        require_ident("cell_id", cell_id)
        require_ident("profile_id", profile_id)
        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=? "
                "AND profile_id=? AND status='queued' ORDER BY created_at LIMIT 1",
                (tenant_id, cell_id, profile_id),
            ).fetchone()
            if not row:
                db.commit()
                return None
            db.execute(
                "UPDATE fabric_jobs SET status='running', updated_at=? WHERE job_id=?",
                (isoformat(), row[0]),
            )
            db.commit()
        return FabricJob(*row[:7])
    def health(self) -> dict[str, Any]:
        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",
            "jobs": count,
            "checked_at": datetime.now(timezone.utc).isoformat(),
        }
@@ -1,47 +0,0 @@
 """Hermes shared fabric profile isolation primitives."""
 from __future__ import annotations
 from dataclasses import dataclass
 from pathlib import Path
 import os
 from .tenant_context import ValidationError, require_ident
@dataclass(frozen=True)
 class HermesProfile:
    tenant_id: str
    cell_id: str
    profile_id: str
    hermes_home: Path
    workspace: Path
    def __post_init__(self) -> None:
        require_ident("tenant_id", self.tenant_id, tenant=True)
        for name in ("cell_id", "profile_id"):
            require_ident(name, getattr(self, name))
        if not self.hermes_home.is_absolute() or not self.workspace.is_absolute():
            raise ValidationError("profile paths must be absolute")
    @property
    def environment(self) -> dict[str, str]:
        return {
            "HERMES_HOME": str(self.hermes_home),
            "HOME": str(self.hermes_home),
            "AIOS_TENANT_ID": self.tenant_id,
            "AIOS_CELL_ID": self.cell_id,
            "AIOS_PROFILE_ID": self.profile_id,
        }
    def prepare(self) -> None:
        self.hermes_home.mkdir(parents=True, exist_ok=True)
        self.workspace.mkdir(parents=True, exist_ok=True)
        os.chmod(self.hermes_home, 0o700)
        os.chmod(self.workspace, 0o700)
    def assert_path_allowed(self, path: str | Path) -> None:
        target = Path(path).resolve()
        for root in (self.hermes_home.resolve(), self.workspace.resolve()):
            if target == root or root in target.parents:
                return
        raise ValidationError("path escapes profile isolation boundary")
@@ -1,158 +0,0 @@
 """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 "",
        )
@@ -1,60 +0,0 @@
 """Tenant-scoped memory namespace and access policy for AIOS V2R1."""
 from __future__ import annotations
 from dataclasses import dataclass
 from pathlib import PurePosixPath
 from typing import Literal
 from .tenant_context import ScopeError, ValidationError, require_ident
 MemoryKind = Literal["session", "profile", "cell", "tenant", "shared_artifact"]
@dataclass(frozen=True)
 class MemoryScope:
    tenant_id: str
    cell_id: str
    profile_id: str
    session_id: str | None
    kind: MemoryKind
    def __post_init__(self) -> None:
        require_ident("tenant_id", self.tenant_id, tenant=True)
        for name in ("cell_id", "profile_id"):
            require_ident(name, getattr(self, name))
        if self.session_id is not None:
            require_ident("session_id", self.session_id)
        if self.kind not in {"session", "profile", "cell", "tenant", "shared_artifact"}:
            raise ValidationError("invalid memory kind")
        if self.kind == "session" and self.session_id is None:
            raise ValidationError("session memory requires session_id")
    @property
    def namespace(self) -> str:
        parts = ["tenant", self.tenant_id, "cell", self.cell_id]
        if self.kind in {"profile", "session"}:
            parts.extend(["profile", self.profile_id])
        if self.kind == "session":
            parts.extend(["session", str(self.session_id)])
        parts.extend(["memory", self.kind])
        return str(PurePosixPath(*parts))
    def assert_access(self, requested: "MemoryScope", *, write: bool = False) -> None:
        if self.tenant_id != requested.tenant_id:
            raise ScopeError("cross-tenant memory access denied")
        if self.cell_id != requested.cell_id:
            raise ScopeError("cross-cell memory access denied")
        if requested.kind == "shared_artifact":
            if write:
                raise ScopeError("shared artifacts are immutable")
            return
        if self.profile_id != requested.profile_id and requested.kind in {"profile", "session"}:
            raise ScopeError("cross-profile memory access denied")
        if requested.kind == "session" and self.session_id != requested.session_id:
            raise ScopeError("cross-session memory access denied")
 def qualify_memory_key(scope: MemoryScope, key: str) -> str:
    key = require_ident("memory_key", key)
    return f"{scope.namespace}/{key}"
@@ -1,224 +0,0 @@
 """Tenant-scoped append-only message ledger and exact quote resolver."""
 from __future__ import annotations
 from dataclasses import dataclass
 from datetime import timedelta
 from contextlib import closing
 import os
 from pathlib import Path
 import sqlite3
 from typing import Iterable
 from .tenant_context import (
    CanonicalMessage,
    ConflictError,
    ScopeError,
    ValidationError,
    isoformat,
    normalize_message_id,
    parse_time,
    utc_now,
 )
@dataclass(frozen=True)
 class LedgerRecord:
    tenant_id: str
    cell_id: str
    channel: str
    channel_account_id: str
    peer_id: str
    direction: str
    message_id: str
    event_id: str
    body: str
    occurred_at: str
    expires_at: str
 class MessageLedger:
    """Exact-ID ledger.
    The resolver never uses time windows, prefixes, similarity, or nearest-message
    fallbacks. Ambiguity and absence fail closed.
    """
    def __init__(self, database: str | Path, *, ttl_days: int = 7) -> None:
        if not 1 <= ttl_days <= 365:
            raise ValidationError("ttl_days must be between 1 and 365")
        self.database = Path(database)
        self.ttl_days = ttl_days
        self.database.parent.mkdir(parents=True, exist_ok=True)
        self._initialize()
        try:
            os.chmod(self.database, 0o600)
        except OSError:
            pass
    def _connect(self) -> sqlite3.Connection:
        db = sqlite3.connect(str(self.database), timeout=30)
        db.row_factory = sqlite3.Row
        db.execute("PRAGMA journal_mode=WAL")
        db.execute("PRAGMA synchronous=FULL")
        db.execute("PRAGMA busy_timeout=30000")
        return db
    def _initialize(self) -> None:
        with closing(self._connect()) as connection, connection as db:
            db.executescript(
                """
                CREATE TABLE IF NOT EXISTS message_ledger (
                    tenant_id TEXT NOT NULL,
                    cell_id TEXT NOT NULL,
                    channel TEXT NOT NULL,
                    channel_account_id TEXT NOT NULL,
                    peer_id TEXT NOT NULL,
                    direction TEXT NOT NULL CHECK(direction IN ('inbound','outbound','internal')),
                    message_id TEXT NOT NULL,
                    event_id TEXT NOT NULL,
                    body TEXT NOT NULL,
                    occurred_at TEXT NOT NULL,
                    expires_at TEXT NOT NULL,
                    created_at TEXT NOT NULL,
                    PRIMARY KEY(
                      tenant_id, cell_id, channel, channel_account_id,
                      peer_id, direction, message_id
                    ),
                    UNIQUE(tenant_id, cell_id, event_id)
                );
                CREATE INDEX IF NOT EXISTS idx_message_reply_lookup ON message_ledger(
                    tenant_id, cell_id, channel, channel_account_id,
                    peer_id, message_id, expires_at
                );
                """
            )
    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))
        key = (
            message.tenant_id,
            message.cell_id,
            message.channel,
            message.channel_account_id,
            message.peer_id,
            message.direction,
            message_id,
        )
        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()
        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
        if message.reply_to_message_id is None:
            raise ScopeError("quoted_text_unavailable")
        reply_id = normalize_message_id(
            message.reply_to_message_id,
            field_name="reply_to_message_id",
        )
        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=? "
                "AND peer_id=? AND message_id=? AND expires_at>?",
                (
                    message.tenant_id,
                    message.cell_id,
                    message.channel,
                    message.channel_account_id,
                    message.peer_id,
                    reply_id,
                    isoformat(),
                ),
            ).fetchall()
        if len(rows) == 1:
            return str(rows[0]["body"])
        if len(rows) > 1:
            raise ConflictError("ambiguous reply ID across message directions")
        raise ScopeError("quoted_text_unavailable")
    def purge_expired(self) -> int:
        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 closing(self._connect()) as connection, connection as db:
            return str(db.execute("PRAGMA integrity_check").fetchone()[0])
    def count(self) -> int:
        with closing(self._connect()) as connection, connection as db:
            return int(db.execute("SELECT count(*) FROM message_ledger").fetchone()[0])
@@ -1,336 +0,0 @@
 """Authoritative tenant registry and fail-closed message router."""
 from __future__ import annotations
 from contextlib import contextmanager
 from contextlib import closing
 from dataclasses import dataclass
 from datetime import datetime, timedelta
 import sqlite3
 from pathlib import Path
 from typing import Callable, Iterator
 from .tenant_context import (
    CanonicalMessage,
    ConflictError,
    ScopeError,
    TenantContext,
    ValidationError,
    isoformat,
    parse_time,
    require_ident,
    utc_now,
 )
@dataclass(frozen=True)
 class Route:
    tenant_id: str
    cell_id: str
    channel: str
    channel_account_id: str
    principal_id: str
    session_id: str | None
 class TenantRouter:
    """SQLite-backed routing registry.
    Channel accounts are authoritative tenant ownership boundaries. Principal and
    session bindings may narrow a route, but can never move traffic to another
    tenant or Cell.
    """
    def __init__(
        self, database: str | Path, *, clock: Callable[[], datetime] = utc_now
    ) -> None:
        self.database = str(database)
        self._clock = clock
        self._initialize()
    def _connect(self) -> sqlite3.Connection:
        connection = sqlite3.connect(self.database, timeout=30, isolation_level=None)
        connection.row_factory = sqlite3.Row
        connection.execute("PRAGMA foreign_keys=ON")
        connection.execute("PRAGMA journal_mode=WAL")
        connection.execute("PRAGMA synchronous=FULL")
        connection.execute("PRAGMA busy_timeout=30000")
        return connection
    @contextmanager
    def _transaction(self) -> Iterator[sqlite3.Connection]:
        connection = self._connect()
        try:
            connection.execute("BEGIN IMMEDIATE")
            yield connection
            connection.execute("COMMIT")
        except Exception:
            connection.execute("ROLLBACK")
            raise
        finally:
            connection.close()
    def _initialize(self) -> None:
        with closing(self._connect()) as connection, connection as db:
            db.executescript(
                """
                CREATE TABLE IF NOT EXISTS tenants (
                    tenant_id TEXT PRIMARY KEY,
                    cell_id TEXT NOT NULL UNIQUE,
                    status TEXT NOT NULL CHECK(status IN ('active','suspended','retired')),
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL
                );
                CREATE TABLE IF NOT EXISTS channel_routes (
                    channel TEXT NOT NULL,
                    channel_account_id TEXT NOT NULL,
                    tenant_id TEXT NOT NULL,
                    cell_id TEXT NOT NULL,
                    created_at TEXT NOT NULL,
                    PRIMARY KEY(channel, channel_account_id),
                    FOREIGN KEY(tenant_id) REFERENCES tenants(tenant_id)
                );
                CREATE TABLE IF NOT EXISTS principal_routes (
                    channel TEXT NOT NULL,
                    channel_account_id TEXT NOT NULL,
                    principal_id TEXT NOT NULL,
                    tenant_id TEXT NOT NULL,
                    cell_id TEXT NOT NULL,
                    created_at TEXT NOT NULL,
                    PRIMARY KEY(channel, channel_account_id, principal_id),
                    FOREIGN KEY(channel, channel_account_id)
                      REFERENCES channel_routes(channel, channel_account_id)
                );
                CREATE TABLE IF NOT EXISTS session_bindings (
                    tenant_id TEXT NOT NULL,
                    session_id TEXT NOT NULL,
                    cell_id TEXT NOT NULL,
                    principal_id TEXT NOT NULL,
                    expires_at TEXT NOT NULL,
                    created_at TEXT NOT NULL,
                    PRIMARY KEY(tenant_id, session_id),
                    FOREIGN KEY(tenant_id) REFERENCES tenants(tenant_id)
                );
                CREATE INDEX IF NOT EXISTS idx_session_expiry
                  ON session_bindings(expires_at);
                """
            )
    def register_tenant(self, tenant_id: str, cell_id: str) -> None:
        tenant_id = require_ident("tenant_id", tenant_id, tenant=True)
        cell_id = require_ident("cell_id", cell_id)
        now = isoformat(self._clock())
        with self._transaction() as db:
            row = db.execute(
                "SELECT cell_id, status FROM tenants WHERE tenant_id=?", (tenant_id,)
            ).fetchone()
            if row:
                if row["cell_id"] != cell_id:
                    raise ConflictError("tenant already bound to another cell")
                db.execute(
                    "UPDATE tenants SET status='active', updated_at=? WHERE tenant_id=?",
                    (now, tenant_id),
                )
                return
            conflict = db.execute(
                "SELECT tenant_id FROM tenants WHERE cell_id=?", (cell_id,)
            ).fetchone()
            if conflict:
                raise ConflictError("cell already belongs to another tenant")
            db.execute(
                "INSERT INTO tenants VALUES(?,?,?,?,?)",
                (tenant_id, cell_id, "active", now, now),
            )
    def set_tenant_status(self, tenant_id: str, status: str) -> None:
        tenant_id = require_ident("tenant_id", tenant_id, tenant=True)
        if status not in {"active", "suspended", "retired"}:
            raise ValidationError("invalid tenant status")
        with self._transaction() as db:
            cursor = db.execute(
                "UPDATE tenants SET status=?, updated_at=? WHERE tenant_id=?",
                (status, isoformat(self._clock()), tenant_id),
            )
            if cursor.rowcount != 1:
                raise ValidationError("unknown tenant")
    def bind_channel(
        self,
        *,
        tenant_id: str,
        channel: str,
        channel_account_id: str,
    ) -> None:
        tenant_id = require_ident("tenant_id", tenant_id, tenant=True)
        channel = require_ident("channel", channel)
        channel_account_id = require_ident("channel_account_id", channel_account_id)
        with self._transaction() as db:
            tenant = db.execute(
                "SELECT cell_id, status FROM tenants WHERE tenant_id=?", (tenant_id,)
            ).fetchone()
            if not tenant or tenant["status"] != "active":
                raise ValidationError("tenant is not active")
            existing = db.execute(
                "SELECT tenant_id, cell_id FROM channel_routes "
                "WHERE channel=? AND channel_account_id=?",
                (channel, channel_account_id),
            ).fetchone()
            if existing:
                if existing["tenant_id"] != tenant_id or existing["cell_id"] != tenant["cell_id"]:
                    raise ConflictError("channel account already belongs to another tenant")
                return
            db.execute(
                "INSERT INTO channel_routes VALUES(?,?,?,?,?)",
                (channel, channel_account_id, tenant_id, tenant["cell_id"], isoformat(self._clock())),
            )
    def bind_principal(
        self,
        *,
        tenant_id: str,
        channel: str,
        channel_account_id: str,
        principal_id: str,
    ) -> None:
        principal_id = require_ident("principal_id", principal_id)
        route = self.resolve_channel(channel, channel_account_id)
        if route.tenant_id != tenant_id:
            raise ScopeError("principal cannot cross tenant boundary")
        with self._transaction() as db:
            existing = db.execute(
                "SELECT tenant_id, cell_id FROM principal_routes WHERE "
                "channel=? AND channel_account_id=? AND principal_id=?",
                (channel, channel_account_id, principal_id),
            ).fetchone()
            if existing:
                if existing["tenant_id"] != tenant_id or existing["cell_id"] != route.cell_id:
                    raise ConflictError("principal route conflict")
                return
            db.execute(
                "INSERT INTO principal_routes VALUES(?,?,?,?,?,?)",
                (
                    channel,
                    channel_account_id,
                    principal_id,
                    tenant_id,
                    route.cell_id,
                    isoformat(self._clock()),
                ),
            )
    def bind_session(
        self,
        *,
        tenant_id: str,
        session_id: str,
        cell_id: str,
        principal_id: str,
        ttl_seconds: int = 86400,
    ) -> None:
        tenant_id = require_ident("tenant_id", tenant_id, tenant=True)
        session_id = require_ident("session_id", session_id)
        cell_id = require_ident("cell_id", cell_id)
        principal_id = require_ident("principal_id", principal_id)
        if not 60 <= ttl_seconds <= 31 * 86400:
            raise ValidationError("invalid session TTL")
        now = self._clock()
        expires = isoformat(now + timedelta(seconds=ttl_seconds))
        with self._transaction() as db:
            tenant = db.execute(
                "SELECT cell_id, status FROM tenants WHERE tenant_id=?", (tenant_id,)
            ).fetchone()
            if not tenant or tenant["status"] != "active" or tenant["cell_id"] != cell_id:
                raise ScopeError("session cell does not match tenant")
            existing = db.execute(
                "SELECT cell_id, principal_id FROM session_bindings "
                "WHERE tenant_id=? AND session_id=?",
                (tenant_id, session_id),
            ).fetchone()
            if existing and (
                existing["cell_id"] != cell_id or existing["principal_id"] != principal_id
            ):
                raise ConflictError("session is already bound to another principal or cell")
            db.execute(
                "INSERT INTO session_bindings VALUES(?,?,?,?,?,?) "
                "ON CONFLICT(tenant_id,session_id) DO UPDATE SET expires_at=excluded.expires_at",
                (tenant_id, session_id, cell_id, principal_id, expires, isoformat(now)),
            )
    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 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 "
                "WHERE r.channel=? AND r.channel_account_id=?",
                (channel, channel_account_id),
            ).fetchone()
        if not row:
            raise ScopeError("unknown channel account")
        if row["status"] != "active":
            raise ScopeError("tenant is not active")
        return Route(row["tenant_id"], row["cell_id"], channel, channel_account_id, "", None)
    def resolve(
        self,
        *,
        channel: str,
        channel_account_id: str,
        principal_id: str,
        session_id: str | None = None,
    ) -> Route:
        principal_id = require_ident("principal_id", principal_id)
        base = self.resolve_channel(channel, channel_account_id)
        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=?",
                (channel, channel_account_id, principal_id),
            ).fetchone()
            if principal and (
                principal["tenant_id"] != base.tenant_id or principal["cell_id"] != base.cell_id
            ):
                raise ScopeError("principal route violates channel ownership")
            if session_id is not None:
                session_id = require_ident("session_id", session_id)
                session = db.execute(
                    "SELECT cell_id, principal_id, expires_at FROM session_bindings "
                    "WHERE tenant_id=? AND session_id=?",
                    (base.tenant_id, session_id),
                ).fetchone()
                if not session:
                    raise ScopeError("unknown session")
                if parse_time(session["expires_at"]) <= self._clock():
                    raise ScopeError("session expired")
                if session["cell_id"] != base.cell_id or session["principal_id"] != principal_id:
                    raise ScopeError("session scope mismatch")
        return Route(
            base.tenant_id,
            base.cell_id,
            channel,
            channel_account_id,
            principal_id,
            session_id,
        )
    def assert_message_route(self, message: CanonicalMessage) -> Route:
        route = self.resolve(
            channel=message.channel,
            channel_account_id=message.channel_account_id,
            principal_id=message.principal_id,
            session_id=message.session_id,
        )
        if message.tenant_id != route.tenant_id or message.cell_id != route.cell_id:
            raise ScopeError("message claims do not match authoritative route")
        return route
    def assert_context_route(self, context: TenantContext) -> Route:
        route = self.resolve(
            channel=context.channel,
            channel_account_id=context.channel_account_id,
            principal_id=context.principal_id,
            session_id=context.session_id,
        )
        if context.tenant_id != route.tenant_id or context.cell_id != route.cell_id:
            raise ScopeError("context claims do not match authoritative route")
        return route
@@ -1,136 +0,0 @@
 """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")
@@ -1,33 +0,0 @@
 """Tenant and Cell scoped secret reference model.
 This module never stores secret values. It stores ownership and retrieval scope
 metadata for integration with a real secret broker.
 """
 from __future__ import annotations
 from dataclasses import dataclass
 from typing import Literal
 from .tenant_context import ValidationError, require_ident
@dataclass(frozen=True)
 class SecretRef:
    secret_id: str
    tenant_id: str
    cell_id: str
    kind: Literal["channel", "model", "tool", "runtime"]
    version: str
    broker_path: str
    def __post_init__(self) -> None:
        for name in ("secret_id", "cell_id", "version"):
            require_ident(name, getattr(self, name))
        require_ident("tenant_id", self.tenant_id, tenant=True)
        if self.kind not in {"channel", "model", "tool", "runtime"}:
            raise ValidationError("invalid secret kind")
        if not self.broker_path.startswith("secret://"):
            raise ValidationError("secret broker path required")
    def can_be_used_by(self, *, tenant_id: str, cell_id: str) -> bool:
        return self.tenant_id == tenant_id and self.cell_id == cell_id
@@ -1,59 +0,0 @@
 """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)
@@ -1,316 +0,0 @@
 """Signed tenant context and canonical message contract for AIOS V2R1."""
 from __future__ import annotations
 import base64
 from dataclasses import asdict, dataclass
 from datetime import datetime, timedelta, timezone
 import hashlib
 import hmac
 import json
 import re
 import secrets
 from typing import Any, Mapping
 VERSION = "V2R1C0"
 MAX_SAFE_INTEGER = 9_007_199_254_740_991
 TENANT_RE = re.compile(r"^[a-z0-9](?:[a-z0-9-]{0,38}[a-z0-9])?$")
 IDENT_RE = re.compile(r"^[A-Za-z0-9](?:[A-Za-z0-9._:@-]{0,126}[A-Za-z0-9])?$")
 class V2R1Error(RuntimeError):
    pass
 class ValidationError(V2R1Error):
    pass
 class ScopeError(V2R1Error):
    pass
 class ConflictError(V2R1Error):
    pass
 def utc_now() -> datetime:
    return datetime.now(timezone.utc)
 def isoformat(value: datetime | None = None) -> str:
    return (value or utc_now()).astimezone(timezone.utc).isoformat()
 def parse_time(value: str) -> datetime:
    if not isinstance(value, str) or not value.strip():
        raise ValidationError("timestamp must be a non-empty string")
    text = value.strip()
    if text.endswith("Z"):
        text = text[:-1] + "+00:00"
    try:
        parsed = datetime.fromisoformat(text)
    except ValueError as exc:
        raise ValidationError(f"invalid timestamp: {value!r}") from exc
    if parsed.tzinfo is None:
        raise ValidationError("timestamp must include timezone")
    return parsed.astimezone(timezone.utc)
 def require_ident(name: str, value: Any, *, tenant: bool = False) -> str:
    if not isinstance(value, str):
        raise ValidationError(f"{name} must be a string")
    value = value.strip()
    pattern = TENANT_RE if tenant else IDENT_RE
    if not pattern.fullmatch(value):
        raise ValidationError(f"invalid {name}: {value!r}")
    return value
 def normalize_message_id(value: Any, *, field_name: str = "message_id") -> str:
    if isinstance(value, bool):
        raise ValidationError(f"{field_name} cannot be boolean")
    if isinstance(value, str):
        value = value.strip()
        if not value or len(value) > 256 or any(ord(ch) < 32 for ch in value):
            raise ValidationError(f"invalid {field_name}")
        return value
    if isinstance(value, int):
        if abs(value) > MAX_SAFE_INTEGER:
            raise ValidationError(f"{field_name} is an unsafe numeric ID")
        return str(value)
    raise ValidationError(f"{field_name} must be a lossless string or safe integer")
 def canonical_json(value: Any) -> bytes:
    return json.dumps(
        value, ensure_ascii=False, separators=(",", ":"), sort_keys=True
    ).encode("utf-8")
 def _b64(data: bytes) -> str:
    return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii")
 def _unb64(value: str) -> bytes:
    return base64.urlsafe_b64decode(value + "=" * (-len(value) % 4))
@dataclass(frozen=True)
 class TenantContext:
    tenant_id: str
    cell_id: str
    principal_id: str
    principal_type: str
    session_id: str
    task_id: str | None
    channel: str
    channel_account_id: str
    issued_at: str
    expires_at: str
    nonce: str
    def __post_init__(self) -> None:
        require_ident("tenant_id", self.tenant_id, tenant=True)
        for name in (
            "cell_id", "principal_id", "principal_type", "session_id",
            "channel", "channel_account_id", "nonce",
        ):
            require_ident(name, getattr(self, name))
        if self.task_id is not None:
            require_ident("task_id", self.task_id)
        issued = parse_time(self.issued_at)
        expires = parse_time(self.expires_at)
        if expires <= issued or expires - issued > timedelta(hours=1):
            raise ValidationError("invalid tenant context lifetime")
    @classmethod
    def issue(cls, *, lifetime_seconds: int = 300, now: datetime | None = None, **scope: Any) -> "TenantContext":
        if not 1 <= lifetime_seconds <= 3600:
            raise ValidationError("lifetime_seconds must be between 1 and 3600")
        current = (now or utc_now()).astimezone(timezone.utc)
        return cls(
            **scope,
            issued_at=isoformat(current),
            expires_at=isoformat(current + timedelta(seconds=lifetime_seconds)),
            nonce=secrets.token_hex(16),
        )
    def sign(self, key: bytes) -> str:
        if len(key) < 32:
            raise ValidationError("signing key must be at least 32 bytes")
        payload = _b64(canonical_json(asdict(self)))
        signature = _b64(hmac.new(key, payload.encode("ascii"), hashlib.sha256).digest())
        return f"{payload}.{signature}"
    @classmethod
    def verify(
        cls,
        token: str,
        key: bytes,
        *,
        now: datetime | None = None,
        expected_tenant_id: str | None = None,
        expected_cell_id: str | None = None,
    ) -> "TenantContext":
        if len(key) < 32:
            raise ValidationError("signing key must be at least 32 bytes")
        try:
            payload, encoded_signature = token.split(".", 1)
            actual = _unb64(encoded_signature)
            if encoded_signature != _b64(actual):
                raise ValidationError("non-canonical tenant context signature")
            data = json.loads(_unb64(payload))
        except Exception as exc:
            raise ValidationError("malformed tenant context token") from exc
        expected = hmac.new(key, payload.encode("ascii"), hashlib.sha256).digest()
        if not hmac.compare_digest(actual, expected):
            raise ValidationError("tenant context signature mismatch")
        context = cls(**data)
        current = (now or utc_now()).astimezone(timezone.utc)
        if current < parse_time(context.issued_at) - timedelta(seconds=60):
            raise ValidationError("tenant context issued in future")
        if current >= parse_time(context.expires_at):
            raise ValidationError("tenant context expired")
        if expected_tenant_id and context.tenant_id != expected_tenant_id:
            raise ScopeError("tenant mismatch")
        if expected_cell_id and context.cell_id != expected_cell_id:
            raise ScopeError("cell mismatch")
        return context
@dataclass(frozen=True)
 class CanonicalMessage:
    tenant_id: str
    cell_id: str
    principal_id: str
    principal_type: str
    origin_type: str
    channel: str
    channel_account_id: str
    peer_id: str
    conversation_id: str
    session_id: str
    task_id: str | None
    event_id: str
    message_id: str
    reply_to_message_id: str | None
    direction: str
    command_capable: bool
    occurred_at: str
    received_at: str
    body: str
    quoted_body: str | None
    adapter_name: str
    adapter_version: str
    schema_version: str = "aios.message.v1"
    def __post_init__(self) -> None:
        if self.schema_version != "aios.message.v1":
            raise ValidationError("unsupported message schema version")
        require_ident("tenant_id", self.tenant_id, tenant=True)
        for name in (
            "cell_id", "principal_id", "principal_type", "origin_type", "channel",
            "channel_account_id", "peer_id", "conversation_id", "session_id",
            "event_id", "adapter_name", "adapter_version",
        ):
            require_ident(name, getattr(self, name))
        if self.task_id is not None:
            require_ident("task_id", self.task_id)
        normalize_message_id(self.message_id)
        if self.reply_to_message_id is not None:
            normalize_message_id(self.reply_to_message_id, field_name="reply_to_message_id")
        if self.direction not in {"inbound", "outbound", "internal"}:
            raise ValidationError("invalid direction")
        if not isinstance(self.command_capable, bool):
            raise ValidationError("command_capable must be boolean")
        parse_time(self.occurred_at)
        parse_time(self.received_at)
        if not isinstance(self.body, str):
            raise ValidationError("body must be a string")
        if self.quoted_body is not None and not isinstance(self.quoted_body, str):
            raise ValidationError("quoted_body must be a string or null")
    @classmethod
    def from_mapping(cls, data: Mapping[str, Any]) -> "CanonicalMessage":
        content = data.get("content") or {}
        adapter = data.get("adapter") or {}
        if not isinstance(content, Mapping) or not isinstance(adapter, Mapping):
            raise ValidationError("content and adapter must be objects")
        values = dict(data)
        return cls(
            schema_version=values.get("schema_version", "aios.message.v1"),
            tenant_id=values["tenant_id"],
            cell_id=values["cell_id"],
            principal_id=values["principal_id"],
            principal_type=values["principal_type"],
            origin_type=values["origin_type"],
            channel=values["channel"],
            channel_account_id=values["channel_account_id"],
            peer_id=values["peer_id"],
            conversation_id=values["conversation_id"],
            session_id=values["session_id"],
            task_id=values.get("task_id"),
            event_id=values["event_id"],
            message_id=normalize_message_id(values["message_id"]),
            reply_to_message_id=(
                normalize_message_id(values["reply_to_message_id"], field_name="reply_to_message_id")
                if values.get("reply_to_message_id") is not None else None
            ),
            direction=values["direction"],
            command_capable=values["command_capable"],
            occurred_at=values["occurred_at"],
            received_at=values["received_at"],
            body=content.get("body", ""),
            quoted_body=content.get("quoted_body"),
            adapter_name=adapter["name"],
            adapter_version=adapter["version"],
        )
    def assert_context(self, context: TenantContext) -> None:
        fields = (
            "tenant_id", "cell_id", "principal_id", "session_id", "task_id",
            "channel", "channel_account_id",
        )
        mismatches = [name for name in fields if getattr(self, name) != getattr(context, name)]
        if mismatches:
            raise ScopeError("message/context mismatch: " + ", ".join(mismatches))
    @property
    def is_authenticated_human_command(self) -> bool:
        return (
            self.direction == "inbound"
            and self.command_capable
            and self.principal_type == "authenticated_user"
            and self.origin_type == "channel_user"
        )
@dataclass(frozen=True)
 class PolicyDecision:
    allowed: bool
    approval_required: bool
    code: str
    reason: str
 READ_ONLY_TOOLS = frozenset({
    "read", "grep", "glob", "find", "ls", "web_search", "web_fetch",
    "memory_search", "sessions_list", "sessions_history", "session_status",
 })
 def evaluate_tool_call(tool_name: str, message: CanonicalMessage) -> PolicyDecision:
    normalized = tool_name.strip().lower() if isinstance(tool_name, str) else ""
    if not normalized:
        return PolicyDecision(False, False, "invalid_tool", "tool name missing")
    if normalized in READ_ONLY_TOOLS:
        return PolicyDecision(True, False, "read_only", "read-only operation")
    if not message.is_authenticated_human_command:
        return PolicyDecision(
            False, False, "origin_denied",
            "side effects require an authenticated current human command",
        )
    return PolicyDecision(
        True, True, "approval_required",
        "authenticated human side effect requires explicit approval",
    )
@@ -1,120 +0,0 @@
 """Architect governance agent responsibilities.
 This agent records requirement changes, maintains versioned human-readable
 requirement/architecture/deployment/test documents, reviews architecture changes,
 validates deployment docs against actual configuration, and produces version bump
 recommendations.
 """
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from typing import Any
 def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()
@dataclass(frozen=True)
 class RequirementChange:
    id: str
    source: str
    actor: str
    requirement: str
    created_at: str = field(default_factory=utc_now)
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
@dataclass(frozen=True)
 class DocumentRefreshPlan:
    requirement_doc: bool
    architecture_doc: bool
    deployment_doc: bool
    test_case_doc: bool
    reason: str
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
@dataclass(frozen=True)
 class ArchitectureReviewDecision:
    status: str
    findings: list[str]
    optimization_recommendations: list[str]
    version_bump: str
    docs_to_refresh: DocumentRefreshPlan
    created_at: str = field(default_factory=utc_now)
    def to_dict(self) -> dict[str, Any]:
        return {
            "status": self.status,
            "findings": self.findings,
            "optimization_recommendations": self.optimization_recommendations,
            "version_bump": self.version_bump,
            "docs_to_refresh": self.docs_to_refresh.to_dict(),
            "created_at": self.created_at,
        }
 class ArchitectGovernanceAgent:
    """Independent architecture governance role."""
    def record_requirement_change(self, source: str, actor: str, requirement: str) -> RequirementChange:
        change_id = "REQ-" + datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S")
        return RequirementChange(change_id, source, actor, requirement)
    def review_architecture_change(self, proposal: dict[str, Any]) -> ArchitectureReviewDecision:
        impact = proposal.get("impact", "medium")
        findings = [
            "requirement_traceability_checked",
            "architecture_consistency_checked",
            "deployment_doc_alignment_checked",
            "test_case_doc_required",
            "rollback_and_owner_gate_checked",
        ]
        recommendations = [
            "refresh user requirements document",
            "refresh architecture design document",
            "refresh engineering deployment document",
            "generate acceptance test case document",
        ]
        version_bump = self.decide_version_bump(proposal)
        return ArchitectureReviewDecision(
            status="review_required" if impact in {"high", "architecture"} else "pass",
            findings=findings,
            optimization_recommendations=recommendations,
            version_bump=version_bump,
            docs_to_refresh=DocumentRefreshPlan(True, True, True, True, "architecture or requirement changed"),
        )
    def decide_version_bump(self, proposal: dict[str, Any]) -> str:
        if proposal.get("breaks_architecture") or proposal.get("major_capability"):
            return "V"
        if proposal.get("new_release_capability") or proposal.get("architecture_change"):
            return "R"
        return "C"
    def validate_deployment_doc(self, deployment_doc: dict[str, Any], actual_config: dict[str, Any]) -> dict[str, Any]:
        expected = set(deployment_doc.get("services", []))
        actual = set(actual_config.get("services", []))
        missing = sorted(expected - actual)
        extra = sorted(actual - expected)
        return {
            "status": "pass" if not missing else "attention_required",
            "missing_services": missing,
            "extra_services": extra,
            "checked_at": utc_now(),
        }
 __all__ = [
    "RequirementChange",
    "DocumentRefreshPlan",
    "ArchitectureReviewDecision",
    "ArchitectGovernanceAgent",
 ]
@@ -1,32 +0,0 @@
 """API for Architect Governance Center."""
 try:
    from fastapi import APIRouter
 except Exception:
    APIRouter = None
 from .architect_governance_dashboard import ArchitectGovernanceDashboard
 dashboard = ArchitectGovernanceDashboard()
 def get_architect_governance_overview():
    return dashboard.overview()
 def build_router():
    if APIRouter is None:
        return None
    router = APIRouter(prefix="/admin/ai-evolution/architect", tags=["architect-governance"])
    @router.get("/overview", operation_id="getArchitectGovernanceOverview")
    def overview():
        return get_architect_governance_overview()
    return router
 router = build_router()
 __all__ = ["router", "get_architect_governance_overview"]
@@ -1,153 +0,0 @@
 """Web UI dashboard model for Architect Governance Agent.
 Shows requirement changes, architecture review findings, document refresh status,
 deployment consistency, test case readiness, weekly research and V/R/C version
 recommendations.
 """
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from typing import Any
 def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()
@dataclass(frozen=True)
 class ArchitectPanelMetric:
    key: str
    title: str
    value: Any
    status: str = "ok"
    detail: str = ""
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
@dataclass(frozen=True)
 class ArchitectPanel:
    key: str
    title: str
    description: str
    metrics: list[ArchitectPanelMetric] = field(default_factory=list)
    def to_dict(self) -> dict[str, Any]:
        return {
            "key": self.key,
            "title": self.title,
            "description": self.description,
            "metrics": [m.to_dict() for m in self.metrics],
        }
 class ArchitectGovernanceDashboard:
    def overview(
        self,
        requirement_changes: list[dict[str, Any]] | None = None,
        architecture_reviews: list[dict[str, Any]] | None = None,
        document_status: dict[str, Any] | None = None,
        research_status: dict[str, Any] | None = None,
        version_recommendations: list[dict[str, Any]] | None = None,
    ) -> dict[str, Any]:
        requirement_changes = requirement_changes or []
        architecture_reviews = architecture_reviews or []
        document_status = document_status or {}
        research_status = research_status or {}
        version_recommendations = version_recommendations or []
        panels = [
            self.requirement_panel(requirement_changes),
            self.architecture_panel(architecture_reviews),
            self.document_panel(document_status),
            self.research_panel(research_status),
            self.version_panel(version_recommendations),
        ]
        return {
            "module": "architect-governance",
            "route": "/admin/ai-evolution/architect",
            "title": "架构师 Agent 治理中心",
            "status": "attention_required" if any(self._panel_warn(panel) for panel in panels) else "healthy",
            "generated_at": utc_now(),
            "panels": [panel.to_dict() for panel in panels],
            "requirement_changes": requirement_changes,
            "architecture_reviews": architecture_reviews,
            "version_recommendations": version_recommendations,
        }
    def requirement_panel(self, changes: list[dict[str, Any]]) -> ArchitectPanel:
        return ArchitectPanel(
            key="requirements",
            title="需求变化记录",
            description="记录来自 Custom GPT、Codex、OpenClaw、Hermes 和用户反馈的需求变化。",
            metrics=[
                ArchitectPanelMetric("total_requirement_changes", "需求变化数量", len(changes)),
                ArchitectPanelMetric(
                    "undocumented_changes",
                    "待刷新到用户需求文档",
                    len([c for c in changes if not c.get("documented")]),
                    "warn" if any(not c.get("documented") for c in changes) else "ok",
                ),
            ],
        )
    def architecture_panel(self, reviews: list[dict[str, Any]]) -> ArchitectPanel:
        pending = [r for r in reviews if r.get("status") not in {"pass", "not_required"}]
        return ArchitectPanel(
            key="architecture_review",
            title="全量架构审视",
            description="展示架构一致性、扩展性、安全边界、可运维性和回滚策略审查。",
            metrics=[
                ArchitectPanelMetric("review_count", "架构审查数量", len(reviews)),
                ArchitectPanelMetric("pending_reviews", "待处理架构审查", len(pending), "warn" if pending else "ok"),
            ],
        )
    def document_panel(self, status: dict[str, Any]) -> ArchitectPanel:
        required_docs = ["requirements", "architecture", "deployment", "test_cases"]
        missing = [doc for doc in required_docs if status.get(doc) not in {"ready", "updated"}]
        return ArchitectPanel(
            key="documents",
            title="文档版本化状态",
            description="用户需求、架构设计、工程部署、测试例文档刷新状态。",
            metrics=[
                ArchitectPanelMetric("requirements_doc", "用户需求文档", status.get("requirements", "unknown"), "warn" if "requirements" in missing else "ok"),
                ArchitectPanelMetric("architecture_doc", "架构设计文档", status.get("architecture", "unknown"), "warn" if "architecture" in missing else "ok"),
                ArchitectPanelMetric("deployment_doc", "工程部署文档", status.get("deployment", "unknown"), "warn" if "deployment" in missing else "ok"),
                ArchitectPanelMetric("test_case_doc", "测试例文档", status.get("test_cases", "unknown"), "warn" if "test_cases" in missing else "ok"),
            ],
        )
    def research_panel(self, status: dict[str, Any]) -> ArchitectPanel:
        overdue = bool(status.get("overdue"))
        return ArchitectPanel(
            key="weekly_research",
            title="每周外部技术情报",
            description="联网搜索 Agent/模型版本、GitHub top 实现、潜在安全影响。",
            metrics=[
                ArchitectPanelMetric("last_research_at", "最近调研时间", status.get("last_research_at", "never"), "warn" if overdue else "ok"),
                ArchitectPanelMetric("findings", "发现数量", status.get("findings", 0)),
                ArchitectPanelMetric("optimization_candidates", "可借鉴优化项", status.get("optimization_candidates", 0)),
            ],
        )
    def version_panel(self, recommendations: list[dict[str, Any]]) -> ArchitectPanel:
        latest = recommendations[-1] if recommendations else {}
        return ArchitectPanel(
            key="version_decision",
            title="V/R/C 版本建议",
            description="由架构师 Agent 判断变更应进入 C、R 或 V。",
            metrics=[
                ArchitectPanelMetric("recommendation_count", "版本建议数量", len(recommendations)),
                ArchitectPanelMetric("latest_recommendation", "最近建议", latest.get("version_bump", "none")),
                ArchitectPanelMetric("reason", "建议原因", latest.get("reason", "")),
            ],
        )
    def _panel_warn(self, panel: ArchitectPanel) -> bool:
        return any(metric.status == "warn" for metric in panel.metrics)
 __all__ = ["ArchitectPanelMetric", "ArchitectPanel", "ArchitectGovernanceDashboard"]
@@ -1,40 +0,0 @@
 """Web UI model for Architect weekly research."""
 from .architect_weekly_research_agent import ArchitectWeeklyResearchAgent
 class ArchitectResearchDashboard:
    def __init__(self, agent=None):
        self.agent = agent or ArchitectWeeklyResearchAgent()
    def overview(self):
        return {
            "module": "architect-weekly-research",
            "route": "/admin/ai-evolution/architect/research",
            "title": "架构师 Agent 技术情报中心",
            "panels": [
                {
                    "title": "Research Targets",
                    "data": self.agent.plan(),
                },
                {
                    "title": "Evaluation",
                    "fields": [
                        "relevance",
                        "impact",
                        "recommendation",
                    ],
                },
                {
                    "title": "Version Decision",
                    "fields": [
                        "C",
                        "R",
                        "V",
                    ],
                },
            ],
        }
 __all__ = ["ArchitectResearchDashboard"]
@@ -1,122 +0,0 @@
 """Architect research timeline dashboard.
 Shows weekly external research findings as a traceable architecture-impact flow:
 research target -> finding -> impact -> recommendation -> version decision ->
 governance event.
 """
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from typing import Any
 def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()
@dataclass(frozen=True)
 class ResearchTimelineStep:
    id: str
    title: str
    step_type: str
    status: str
    source: str
    timestamp: str
    impact: str = "none"
    version_bump: str = "none"
    detail: dict[str, Any] = field(default_factory=dict)
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
 class ArchitectResearchTimeline:
    ORDER = {
        "research_planned": 10,
        "finding_created": 20,
        "impact_analyzed": 30,
        "recommendation_created": 40,
        "version_decision_created": 50,
        "architecture_review_event_created": 60,
    }
    def build(self, findings: list[dict[str, Any]] | None = None, decisions: list[dict[str, Any]] | None = None) -> dict[str, Any]:
        findings = findings or []
        decisions = decisions or []
        steps: list[ResearchTimelineStep] = [
            ResearchTimelineStep(
                id="research-plan",
                title="每周技术情报调研计划",
                step_type="research_planned",
                status="ok",
                source="architect-agent",
                timestamp=utc_now(),
                detail={"cadence": "weekly"},
            )
        ]
        for idx, finding in enumerate(findings):
            steps.append(
                ResearchTimelineStep(
                    id=finding.get("id", f"finding-{idx}"),
                    title=finding.get("title", "research finding"),
                    step_type="finding_created",
                    status="ok",
                    source=finding.get("source", "unknown"),
                    timestamp=finding.get("created_at", utc_now()),
                    impact=finding.get("impact", "none"),
                    detail=finding,
                )
            )
            steps.append(
                ResearchTimelineStep(
                    id=f"impact-{finding.get('id', idx)}",
                    title="架构影响分析",
                    step_type="impact_analyzed",
                    status="attention_required" if finding.get("impact") in {"architecture", "breaking", "new_capability"} else "ok",
                    source="architect-agent",
                    timestamp=utc_now(),
                    impact=finding.get("impact", "none"),
                    detail={"relevance": finding.get("relevance"), "recommendation": finding.get("recommendation")},
                )
            )
        for idx, decision in enumerate(decisions):
            steps.append(
                ResearchTimelineStep(
                    id=decision.get("id", f"decision-{idx}"),
                    title="C/R/V 版本建议",
                    step_type="version_decision_created",
                    status=decision.get("status", "ok"),
                    source="architect-agent",
                    timestamp=decision.get("created_at", utc_now()),
                    version_bump=decision.get("version_bump", "none"),
                    detail=decision,
                )
            )
        steps.sort(key=lambda s: (self.ORDER.get(s.step_type, 999), s.timestamp))
        return {
            "module": "architect-research-timeline",
            "route": "/admin/ai-evolution/architect/research/timeline",
            "title": "架构师技术情报时间线",
            "status": self._status(steps),
            "generated_at": utc_now(),
            "steps": [step.to_dict() for step in steps],
            "summary": self._summary(steps),
        }
    def _status(self, steps: list[ResearchTimelineStep]) -> str:
        if any(step.status == "attention_required" for step in steps):
            return "attention_required"
        return "healthy"
    def _summary(self, steps: list[ResearchTimelineStep]) -> dict[str, Any]:
        return {
            "findings": len([s for s in steps if s.step_type == "finding_created"]),
            "architecture_impacts": len([s for s in steps if s.impact in {"architecture", "breaking", "new_capability"}]),
            "version_decisions": len([s for s in steps if s.step_type == "version_decision_created"]),
            "recommended_bumps": sorted({s.version_bump for s in steps if s.version_bump != "none"}),
        }
 __all__ = ["ResearchTimelineStep", "ArchitectResearchTimeline"]
@@ -1,32 +0,0 @@
 """API for architect research timeline."""
 try:
    from fastapi import APIRouter
 except Exception:
    APIRouter = None
 from .architect_research_timeline import ArchitectResearchTimeline
 timeline = ArchitectResearchTimeline()
 def get_architect_research_timeline():
    return timeline.build()
 def build_router():
    if APIRouter is None:
        return None
    router = APIRouter(prefix="/admin/ai-evolution/architect/research", tags=["architect-research"])
    @router.get("/timeline", operation_id="getArchitectResearchTimeline")
    def get_timeline():
        return get_architect_research_timeline()
    return router
 router = build_router()
 __all__ = ["router", "get_architect_research_timeline"]
@@ -1,113 +0,0 @@
 """Weekly external research loop for Architect Governance Agent.
 The agent plans and normalizes weekly research tasks. Actual internet access is
 provided by the runtime/search worker; this module defines the audited workflow,
 findings, impact analysis, and V/R/C recommendations.
 """
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from typing import Any
 from uuid import uuid4
 def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()
@dataclass(frozen=True)
 class ResearchTarget:
    id: str
    topic: str
    source_type: str
    query: str
    cadence: str = "weekly"
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
@dataclass(frozen=True)
 class ResearchFinding:
    id: str
    target_id: str
    title: str
    summary: str
    source: str
    relevance: str
    impact: str
    recommendation: str
    created_at: str = field(default_factory=utc_now)
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
@dataclass(frozen=True)
 class ResearchDecision:
    status: str
    version_bump: str
    reason: str
    findings: list[ResearchFinding]
    created_at: str = field(default_factory=utc_now)
    def to_dict(self) -> dict[str, Any]:
        return {
            "status": self.status,
            "version_bump": self.version_bump,
            "reason": self.reason,
            "findings": [f.to_dict() for f in self.findings],
            "created_at": self.created_at,
        }
 class ArchitectWeeklyResearchAgent:
    DEFAULT_TARGETS = (
        ResearchTarget("agent-frameworks", "Agent framework versions", "web", "latest multi-agent orchestration framework releases"),
        ResearchTarget("model-ecosystem", "Model ecosystem changes", "web", "latest LLM model releases tool calling agent features"),
        ResearchTarget("github-top-agentops", "GitHub top AgentOps implementations", "github", "top repositories agent observability multi agent orchestration"),
        ResearchTarget("security-agent-risks", "Agent security changes", "web", "latest AI agent security sandbox prompt injection tool use risks"),
    )
    def plan(self) -> dict[str, Any]:
        return {
            "status": "planned",
            "cadence": "weekly",
            "targets": [target.to_dict() for target in self.DEFAULT_TARGETS],
            "created_at": utc_now(),
        }
    def normalize_finding(
        self,
        target_id: str,
        title: str,
        summary: str,
        source: str,
        relevance: str = "medium",
        impact: str = "none",
        recommendation: str = "monitor",
    ) -> ResearchFinding:
        return ResearchFinding(
            id=f"finding_{uuid4().hex[:12]}",
            target_id=target_id,
            title=title,
            summary=summary,
            source=source,
            relevance=relevance,
            impact=impact,
            recommendation=recommendation,
        )
    def decide(self, findings: list[ResearchFinding]) -> ResearchDecision:
        if any(f.impact == "breaking" for f in findings):
            return ResearchDecision("action_required", "V", "breaking external ecosystem change", findings)
        if any(f.impact in {"architecture", "new_capability"} for f in findings):
            return ResearchDecision("optimization_recommended", "R", "external capability may improve architecture", findings)
        if any(f.impact in {"bugfix", "minor"} for f in findings):
            return ResearchDecision("minor_change_recommended", "C", "minor compatible improvement", findings)
        return ResearchDecision("no_action", "none", "no relevant external change", findings)
 __all__ = ["ResearchTarget", "ResearchFinding", "ResearchDecision", "ArchitectWeeklyResearchAgent"]
@@ -1,32 +0,0 @@
 """API for architecture change governance dashboard."""
 try:
    from fastapi import APIRouter
 except Exception:
    APIRouter = None
 from .architecture_change_dashboard import ArchitectureChangeDashboard
 dashboard = ArchitectureChangeDashboard()
 def get_architecture_change_overview():
    return dashboard.overview()
 def build_router():
    if APIRouter is None:
        return None
    router = APIRouter(prefix="/admin/ai-evolution/architecture-changes", tags=["architecture-changes"])
    @router.get("/overview", operation_id="getArchitectureChangeOverview")
    def overview():
        return get_architecture_change_overview()
    return router
 router = build_router()
 __all__ = ["router", "get_architecture_change_overview"]
@@ -1,90 +0,0 @@
 """Architecture change classification.
 Small compatible architecture changes may be handled automatically by the
 Architect Agent with audit evidence. Large, risky, production-impacting or
 breaking architecture changes require explicit Owner approval before
 implementation.
 """
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from typing import Any
@dataclass(frozen=True)
 class ArchitectureChangeClassification:
    level: str
    owner_required: bool
    auto_implement_allowed: bool
    version_bump: str
    reasons: list[str] = field(default_factory=list)
    required_gates: list[str] = field(default_factory=list)
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
 class ArchitectureChangeClassifier:
    """Classifies architecture changes into small/medium/large governance paths."""
    LARGE_FLAGS = {
        "breaking_change",
        "breaks_architecture",
        "production_architecture_update",
        "security_boundary_change",
        "cross_tenant_effect",
        "cross_user_data_effect",
        "data_migration",
        "permission_model_change",
        "runtime_infrastructure_change",
    }
    def classify(self, change: dict[str, Any]) -> ArchitectureChangeClassification:
        risk = str(change.get("risk", "medium")).lower()
        version_bump = str(change.get("version_bump", change.get("recommended_version_bump", "C"))).upper()
        scope = str(change.get("scope", "component")).lower()
        reasons: list[str] = []
        for flag in sorted(self.LARGE_FLAGS):
            if change.get(flag):
                reasons.append(flag)
        if risk in {"high", "critical"}:
            reasons.append(f"risk={risk}")
        if version_bump == "V":
            reasons.append("version_bump=V")
        if version_bump == "R" and scope in {"system", "platform", "production", "cross_agent", "cross_tenant"}:
            reasons.append(f"version_bump=R scope={scope}")
        if reasons:
            return ArchitectureChangeClassification(
                level="large",
                owner_required=True,
                auto_implement_allowed=False,
                version_bump=version_bump,
                reasons=reasons,
                required_gates=["architect_review", "owner_architecture_approval", "test_gate", "promotion_gate"],
            )
        if risk == "low" and version_bump in {"C", "NONE"} and scope in {"component", "local", "documentation", "test"}:
            return ArchitectureChangeClassification(
                level="small",
                owner_required=False,
                auto_implement_allowed=True,
                version_bump=version_bump,
                reasons=["low_risk_compatible_change"],
                required_gates=["architect_review", "audit", "test_gate"],
            )
        return ArchitectureChangeClassification(
            level="medium",
            owner_required=False,
            auto_implement_allowed=False,
            version_bump=version_bump,
            reasons=["requires_architect_review_before_implementation"],
            required_gates=["architect_review", "test_gate", "audit"],
        )
 __all__ = ["ArchitectureChangeClassification", "ArchitectureChangeClassifier"]
@@ -1,93 +0,0 @@
 """Architecture change governance dashboard.
 Displays architecture change classification, owner approval requirements,
 implementation readiness, version recommendation and audit evidence for Web UI.
 """
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from typing import Any
 from .architecture_change_classifier import ArchitectureChangeClassifier
 def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()
@dataclass(frozen=True)
 class ArchitectureChangeView:
    change_id: str
    title: str
    level: str
    owner_required: bool
    auto_implement_allowed: bool
    version_bump: str
    status: str
    reasons: list[str] = field(default_factory=list)
    required_gates: list[str] = field(default_factory=list)
    owner_decision: str | None = None
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
 class ArchitectureChangeDashboard:
    def __init__(self, classifier: ArchitectureChangeClassifier | None = None):
        self.classifier = classifier or ArchitectureChangeClassifier()
    def overview(self, changes: list[dict[str, Any]] | None = None) -> dict[str, Any]:
        changes = changes or []
        views = [self._view(change) for change in changes]
        return {
            "module": "architecture-change-governance",
            "route": "/admin/ai-evolution/architecture-changes",
            "title": "架构变更治理中心",
            "status": self._status(views),
            "generated_at": utc_now(),
            "summary": self._summary(views),
            "changes": [view.to_dict() for view in views],
        }
    def _view(self, change: dict[str, Any]) -> ArchitectureChangeView:
        classification = self.classifier.classify(change)
        owner_decision = change.get("owner_decision")
        if classification.owner_required and owner_decision != "approved":
            status = "waiting_owner_approval"
        elif classification.auto_implement_allowed:
            status = "auto_implementation_allowed"
        else:
            status = "ready_after_required_gates"
        return ArchitectureChangeView(
            change_id=change.get("change_id", change.get("id", "unknown")),
            title=change.get("title", "architecture change"),
            level=classification.level,
            owner_required=classification.owner_required,
            auto_implement_allowed=classification.auto_implement_allowed,
            version_bump=classification.version_bump,
            status=status,
            reasons=classification.reasons,
            required_gates=classification.required_gates,
            owner_decision=owner_decision,
        )
    def _status(self, views: list[ArchitectureChangeView]) -> str:
        if any(view.status == "waiting_owner_approval" for view in views):
            return "attention_required"
        return "healthy"
    def _summary(self, views: list[ArchitectureChangeView]) -> dict[str, Any]:
        return {
            "total": len(views),
            "small": len([v for v in views if v.level == "small"]),
            "medium": len([v for v in views if v.level == "medium"]),
            "large": len([v for v in views if v.level == "large"]),
            "waiting_owner_approval": len([v for v in views if v.status == "waiting_owner_approval"]),
            "auto_allowed": len([v for v in views if v.auto_implement_allowed]),
            "version_bumps": sorted({v.version_bump for v in views if v.version_bump and v.version_bump != "NONE"}),
        }
 __all__ = ["ArchitectureChangeView", "ArchitectureChangeDashboard"]
@@ -1,89 +0,0 @@
 """Owner approval gate for architecture changes.
 Architecture changes may be proposed, reviewed and optimized by agents, but they
 must not be implemented until owner approval is recorded. This gate sits before
 implementation and before promotion.
 """
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from typing import Any
 def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()
@dataclass(frozen=True)
 class ArchitectureOwnerApprovalRequest:
    request_id: str
    change_id: str
    title: str
    reason: str
    risk: str
    architect_decision: str
    status: str = "pending_owner_approval"
    created_at: str = field(default_factory=utc_now)
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
@dataclass(frozen=True)
 class ArchitectureOwnerDecision:
    request_id: str
    change_id: str
    decision: str
    decided_by: str
    reason: str = ""
    decided_at: str = field(default_factory=utc_now)
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
 class ArchitectureOwnerGate:
    """Blocks architecture implementation until owner explicitly approves."""
    def create_request(self, change: dict[str, Any], architect_decision: dict[str, Any]) -> ArchitectureOwnerApprovalRequest:
        return ArchitectureOwnerApprovalRequest(
            request_id=f"ARCH-OWNER-{change.get('change_id', change.get('id', 'unknown'))}",
            change_id=change.get("change_id", change.get("id", "unknown")),
            title=change.get("title", "architecture change"),
            reason=change.get("reason", "architecture impact requires owner approval"),
            risk=change.get("risk", "medium"),
            architect_decision=architect_decision.get("status", "reviewed"),
        )
    def can_implement(self, request: dict[str, Any], decision: dict[str, Any] | None = None) -> dict[str, Any]:
        if not decision:
            return {
                "allowed": False,
                "status": "waiting_owner_approval",
                "reason": "architecture change requires owner approval before implementation",
                "request": request,
            }
        if decision.get("decision") == "approved":
            return {
                "allowed": True,
                "status": "owner_approved",
                "reason": decision.get("reason", "approved by owner"),
                "request": request,
                "decision": decision,
            }
        return {
            "allowed": False,
            "status": "owner_rejected",
            "reason": decision.get("reason", "rejected by owner"),
            "request": request,
            "decision": decision,
        }
 __all__ = [
    "ArchitectureOwnerApprovalRequest",
    "ArchitectureOwnerDecision",
    "ArchitectureOwnerGate",
 ]
@@ -1,79 +0,0 @@
 """HTTP handler and display data for architecture owner gate."""
 from __future__ import annotations
 from dataclasses import asdict
 from datetime import datetime, timezone
 from typing import Any
 from .architecture_owner_gate import ArchitectureOwnerGate, ArchitectureOwnerApprovalRequest
 gate = ArchitectureOwnerGate()
 # In-memory store for pending approval requests
 _pending_requests: dict[str, dict[str, Any]] = {}
 _decisions: dict[str, dict[str, Any]] = {}
 def create_owner_approval_request(change: dict[str, Any], architect_decision: dict[str, Any]) -> dict[str, Any]:
    req = gate.create_request(change, architect_decision)
    d = asdict(req)
    _pending_requests[req.change_id] = d
    return d
 def approve_request(change_id: str, decided_by: str = "owner", reason: str = "") -> dict[str, Any]:
    req = _pending_requests.get(change_id)
    if not req:
        return {"error": f"no pending request for change_id={change_id}"}
    decision = asdict(
        ArchitectureOwnerApprovalRequest.__bases__[1].__new__(type("_ArchitectureOwnerDecision", (), {}))  # noqa
    )
    try:
        from .architecture_owner_gate import ArchitectureOwnerDecision
        dec = ArchitectureOwnerDecision(
            request_id=req["request_id"],
            change_id=change_id,
            decision="approved",
            decided_by=decided_by,
            reason=reason,
        )
        d = asdict(dec)
        _decisions[change_id] = d
        # remove from pending
        _pending_requests.pop(change_id, None)
        return {"allowed": True, "decision": d}
    except Exception as exc:
        return {"error": str(exc)}
 def reject_request(change_id: str, decided_by: str = "owner", reason: str = "") -> dict[str, Any]:
    from .architecture_owner_gate import ArchitectureOwnerDecision
    req = _pending_requests.get(change_id)
    if not req:
        return {"error": f"no pending request for change_id={change_id}"}
    try:
        dec = ArchitectureOwnerDecision(
            request_id=req["request_id"],
            change_id=change_id,
            decision="rejected",
            decided_by=decided_by,
            reason=reason,
        )
        d = asdict(dec)
        _decisions[change_id] = d
        _pending_requests.pop(change_id, None)
        return {"allowed": False, "decision": d}
    except Exception as exc:
        return {"error": str(exc)}
 def get_owner_gate_overview() -> dict[str, Any]:
    return {
        "pending_requests": list(_pending_requests.values()),
        "decisions": list(_decisions.values()),
        "pending_count": len(_pending_requests),
        "approved_count": sum(1 for d in _decisions.values() if d.get("decision") == "approved"),
        "rejected_count": sum(1 for d in _decisions.values() if d.get("decision") == "rejected"),
    }
@@ -1,40 +0,0 @@
 """Independent architecture optimization agent."""
 from dataclasses import dataclass, asdict
 from datetime import datetime, timezone
 def utc_now():
    return datetime.now(timezone.utc).isoformat()
@dataclass(frozen=True)
 class ArchitectureReview:
    agent: str
    decision: str
    findings: list[str]
    created_at: str = utc_now()
    def to_dict(self):
        return asdict(self)
 class ArchitectureReviewAgent:
    """Reviews architecture changes from system-wide optimal perspective."""
    def review(self, proposal: dict) -> ArchitectureReview:
        findings = [
            "dependency_impact_checked",
            "security_boundary_checked",
            "scalability_checked",
            "operability_checked",
            "rollback_strategy_checked",
        ]
        return ArchitectureReview(
            "architecture-optimization-agent",
            "pass" if proposal.get("architecture_change") else "not_required",
            findings,
        )
 __all__ = ["ArchitectureReviewAgent", "ArchitectureReview"]
@@ -1,103 +0,0 @@
 """Change acceptance governance for AIOS.
 Every feature change must provide matching tests. Every architecture change must
 be reviewed by an independent architecture agent. User-facing changes must be
 validated by an independent user acceptance agent.
 """
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from typing import Any
 def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()
@dataclass(frozen=True)
 class ChangeDescriptor:
    change_id: str
    change_type: str
    title: str
    files_changed: list[str]
    tests: list[str] = field(default_factory=list)
    user_facing: bool = False
    architecture_impact: bool = False
    risk: str = "low"
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
@dataclass(frozen=True)
 class AcceptanceFinding:
    check: str
    status: str
    detail: str
    required: bool = True
    def passed(self) -> bool:
        return self.status == "pass" or not self.required
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
@dataclass(frozen=True)
 class AcceptanceDecision:
    status: str
    findings: list[AcceptanceFinding]
    decided_at: str = field(default_factory=utc_now)
    def to_dict(self) -> dict[str, Any]:
        return {
            "status": self.status,
            "findings": [f.to_dict() for f in self.findings],
            "decided_at": self.decided_at,
        }
 class ChangeAcceptanceGate:
    def evaluate(self, change: ChangeDescriptor) -> AcceptanceDecision:
        findings: list[AcceptanceFinding] = []
        findings.append(self._tests_required(change))
        findings.append(self._user_acceptance_required(change))
        findings.append(self._architecture_review_required(change))
        findings.append(self._risk_gate(change))
        status = "pass" if all(f.passed() for f in findings) else "blocked"
        return AcceptanceDecision(status, findings)
    def _tests_required(self, change: ChangeDescriptor) -> AcceptanceFinding:
        if change.tests:
            return AcceptanceFinding("tests_present", "pass", "matching tests declared")
        return AcceptanceFinding("tests_present", "fail", "every feature change requires tests")
    def _user_acceptance_required(self, change: ChangeDescriptor) -> AcceptanceFinding:
        if not change.user_facing:
            return AcceptanceFinding("user_acceptance", "pass", "not user-facing")
        has_user_test = any("user" in t.lower() or "acceptance" in t.lower() for t in change.tests)
        return AcceptanceFinding(
            "user_acceptance",
            "pass" if has_user_test else "fail",
            "user-facing change requires user acceptance test",
        )
    def _architecture_review_required(self, change: ChangeDescriptor) -> AcceptanceFinding:
        if not change.architecture_impact:
            return AcceptanceFinding("architecture_review", "pass", "no architecture impact")
        has_arch_test = any("arch" in t.lower() or "architecture" in t.lower() for t in change.tests)
        return AcceptanceFinding(
            "architecture_review",
            "pass" if has_arch_test else "fail",
            "architecture change requires independent architect review artifact/test",
        )
    def _risk_gate(self, change: ChangeDescriptor) -> AcceptanceFinding:
        if change.risk in {"low", "medium", "high"}:
            return AcceptanceFinding("risk_classified", "pass", f"risk={change.risk}")
        return AcceptanceFinding("risk_classified", "fail", "risk must be low/medium/high")
 __all__ = ["ChangeDescriptor", "AcceptanceFinding", "AcceptanceDecision", "ChangeAcceptanceGate"]
@@ -1,34 +0,0 @@
 """API for AIOS change governance dashboard."""
 from typing import Any
 try:
    from fastapi import APIRouter
 except Exception:
    APIRouter = None
 from .change_governance_dashboard import ChangeGovernanceDashboard
 dashboard = ChangeGovernanceDashboard()
 def get_change_governance_overview(changes: list[dict[str, Any]] | None = None):
    return dashboard.overview(changes)
 def build_router():
    if APIRouter is None:
        return None
    router = APIRouter(prefix="/admin/ai-evolution/governance", tags=["change-governance"])
    @router.get("/overview", operation_id="getChangeGovernanceOverview")
    def overview():
        return get_change_governance_overview()
    return router
 router = build_router()
 __all__ = ["router", "get_change_governance_overview"]
@@ -1,93 +0,0 @@
 """Web UI dashboard model for AIOS change governance.
 Shows test gates, user acceptance, architecture review, risk gates and release
 blocking status for human operators and owner review.
 """
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from typing import Any
 def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()
@dataclass(frozen=True)
 class GovernanceMetric:
    key: str
    title: str
    value: Any
    status: str = "ok"
    detail: str = ""
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
@dataclass(frozen=True)
 class GovernancePanel:
    key: str
    title: str
    metrics: list[GovernanceMetric] = field(default_factory=list)
    def to_dict(self) -> dict[str, Any]:
        return {
            "key": self.key,
            "title": self.title,
            "metrics": [m.to_dict() for m in self.metrics],
        }
 class ChangeGovernanceDashboard:
    def overview(self, changes: list[dict[str, Any]] | None = None) -> dict[str, Any]:
        changes = changes or []
        blocked = [c for c in changes if c.get("status") == "blocked"]
        waiting_arch = [c for c in changes if c.get("architecture_review") == "required"]
        waiting_user = [c for c in changes if c.get("user_acceptance") == "required"]
        missing_tests = [c for c in changes if not c.get("tests")]
        panels = [
            GovernancePanel(
                "test_gate",
                "测试 Gate",
                [
                    GovernanceMetric("total_changes", "变更总数", len(changes)),
                    GovernanceMetric("missing_tests", "缺少测试的变更", len(missing_tests), "warn" if missing_tests else "ok"),
                ],
            ),
            GovernancePanel(
                "user_acceptance",
                "用户验收 Agent",
                [
                    GovernanceMetric("waiting_user_acceptance", "等待用户验收", len(waiting_user), "warn" if waiting_user else "ok"),
                ],
            ),
            GovernancePanel(
                "architecture_review",
                "架构师 Agent",
                [
                    GovernanceMetric("waiting_architecture_review", "等待架构审查", len(waiting_arch), "warn" if waiting_arch else "ok"),
                ],
            ),
            GovernancePanel(
                "release_gate",
                "发布 Gate",
                [
                    GovernanceMetric("blocked_changes", "阻塞发布的变更", len(blocked), "warn" if blocked else "ok"),
                ],
            ),
        ]
        return {
            "module": "change-governance",
            "route": "/admin/ai-evolution/governance",
            "title": "AIOS 变更治理中心",
            "status": "attention_required" if blocked or missing_tests or waiting_arch or waiting_user else "healthy",
            "generated_at": utc_now(),
            "panels": [p.to_dict() for p in panels],
            "changes": changes,
        }
 __all__ = ["GovernanceMetric", "GovernancePanel", "ChangeGovernanceDashboard"]
@@ -1,36 +0,0 @@
 """Git repository location resolver with validation.
 Prevents hard-coded repository names from silently pointing to stale paths.
 """
 from __future__ import annotations
 import os
 from pathlib import Path
 class GitRepoLocator:
    def __init__(self, root: str | None = None):
        self.root = Path(root or os.getenv("AIOS_GITOPS_ROOT", "/data/aios-gitops/runtime/gitea-bin/data/repositories"))
    def candidates(self, owner: str = "aios", names: tuple[str, ...] = ("aios.git", "aios-v23.git")):
        return [self.root / owner / name for name in names]
    def locate(self, owner: str = "aios") -> dict:
        found = [str(path) for path in self.candidates(owner) if path.exists()]
        return {
            "root": str(self.root),
            "owner": owner,
            "found": found,
            "healthy": len(found) == 1,
            "warning": "multiple repositories found" if len(found) > 1 else None,
        }
    def require(self, owner: str = "aios") -> Path:
        result = self.locate(owner)
        if not result["healthy"]:
            raise RuntimeError(f"Git repo resolution failed: {result}")
        return Path(result["found"][0])
 __all__ = ["GitRepoLocator"]
@@ -1,148 +0,0 @@
 """Bridge governance agents into the unified Agent Event Bus.
 This module emits normalized governance events for requirement changes,
 architecture reviews, user acceptance, test gates and version decisions. It uses
 duck typing so it can work with the unified AgentEventBus when available, while
 remaining testable without a running bus.
 """
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from typing import Any, Protocol
 from uuid import uuid4
 def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()
 class EventPublisher(Protocol):
    def publish(
        self,
        kind: str,
        source: str,
        target: str,
        summary: str,
        payload: dict[str, Any],
        correlation_id: str | None = None,
    ) -> dict[str, Any]:
        ...
@dataclass(frozen=True)
 class GovernanceEventEnvelope:
    id: str
    kind: str
    source: str
    target: str
    summary: str
    payload: dict[str, Any]
    correlation_id: str | None = None
    created_at: str = field(default_factory=utc_now)
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
 class InMemoryGovernanceEventBus:
    """Small fallback bus for tests and offline governance runs."""
    def __init__(self) -> None:
        self.events: list[dict[str, Any]] = []
    def publish(
        self,
        kind: str,
        source: str,
        target: str,
        summary: str,
        payload: dict[str, Any],
        correlation_id: str | None = None,
    ) -> dict[str, Any]:
        event = GovernanceEventEnvelope(
            id=f"evt_{uuid4().hex[:12]}",
            kind=kind,
            source=source,
            target=target,
            summary=summary,
            payload=payload,
            correlation_id=correlation_id,
        ).to_dict()
        self.events.append(event)
        return event
 class GovernanceEventBridge:
    """Publishes governance lifecycle events to the Agent Event Bus."""
    def __init__(self, bus: EventPublisher | None = None):
        self.bus = bus or InMemoryGovernanceEventBus()
    def requirement_changed(self, requirement: dict[str, Any]) -> dict[str, Any]:
        return self.bus.publish(
            kind="requirement_changed",
            source=requirement.get("source", "unknown"),
            target="architect-agent",
            summary=requirement.get("requirement", "requirement changed"),
            payload=requirement,
            correlation_id=requirement.get("id"),
        )
    def architecture_review_started(self, proposal: dict[str, Any]) -> dict[str, Any]:
        return self.bus.publish(
            kind="architecture_review_started",
            source="architect-agent",
            target="architecture-review-agent",
            summary=proposal.get("title", "architecture review started"),
            payload=proposal,
            correlation_id=proposal.get("change_id"),
        )
    def architecture_review_completed(self, decision: dict[str, Any]) -> dict[str, Any]:
        return self.bus.publish(
            kind="architecture_review_completed",
            source="architecture-review-agent",
            target="change-governance",
            summary=f"architecture review {decision.get('status', 'completed')}",
            payload=decision,
            correlation_id=decision.get("change_id"),
        )
    def user_acceptance_completed(self, report: dict[str, Any]) -> dict[str, Any]:
        return self.bus.publish(
            kind="user_acceptance_completed",
            source="user-acceptance-agent",
            target="change-governance",
            summary=f"user acceptance {report.get('decision', 'completed')}",
            payload=report,
            correlation_id=report.get("change_id"),
        )
    def test_gate_completed(self, result: dict[str, Any]) -> dict[str, Any]:
        return self.bus.publish(
            kind="test_gate_completed",
            source="test-gate",
            target="change-governance",
            summary=f"test gate {result.get('status', 'completed')}",
            payload=result,
            correlation_id=result.get("change_id"),
        )
    def version_decision_created(self, decision: dict[str, Any]) -> dict[str, Any]:
        return self.bus.publish(
            kind="version_decision_created",
            source="architect-agent",
            target="owner-gate",
            summary=f"version bump recommendation: {decision.get('version_bump', 'unknown')}",
            payload=decision,
            correlation_id=decision.get("change_id"),
        )
 __all__ = [
    "GovernanceEventEnvelope",
    "InMemoryGovernanceEventBus",
    "GovernanceEventBridge",
 ]
@@ -1,32 +0,0 @@
 """API for governance timeline visualization."""
 try:
    from fastapi import APIRouter
 except Exception:
    APIRouter = None
 from .governance_timeline_dashboard import GovernanceTimelineDashboard
 dashboard = GovernanceTimelineDashboard()
 def get_governance_timeline(events=None):
    return dashboard.build(events)
 def build_router():
    if APIRouter is None:
        return None
    router = APIRouter(prefix="/admin/ai-evolution/governance", tags=["governance-timeline"])
    @router.get("/timeline", operation_id="getGovernanceTimeline")
    def timeline():
        return get_governance_timeline()
    return router
 router = build_router()
 __all__ = ["router", "get_governance_timeline"]
@@ -1,106 +0,0 @@
 """Governance timeline dashboard.
 Turns governance events into a human-readable flow for Web UI so operators can
 see why a change is blocked or ready for promotion.
 """
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from typing import Any
 def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()
 EVENT_ORDER = {
    "requirement_changed": 10,
    "architecture_review_started": 20,
    "architecture_review_completed": 30,
    "test_gate_completed": 40,
    "user_acceptance_completed": 50,
    "version_decision_created": 60,
    "owner_gate_completed": 70,
    "promotion_completed": 80,
 }
@dataclass(frozen=True)
 class GovernanceTimelineStep:
    id: str
    kind: str
    title: str
    actor: str
    status: str
    timestamp: str
    correlation_id: str | None = None
    detail: dict[str, Any] = field(default_factory=dict)
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
 class GovernanceTimelineDashboard:
    def build(self, events: list[dict[str, Any]] | None = None) -> dict[str, Any]:
        events = events or []
        steps = [self._event_to_step(event) for event in events]
        steps.sort(key=lambda step: (EVENT_ORDER.get(step.kind, 999), step.timestamp))
        blockers = [step for step in steps if step.status in {"blocked", "fail", "attention_required"}]
        return {
            "module": "governance-timeline",
            "route": "/admin/ai-evolution/governance/timeline",
            "title": "治理事件时间线",
            "status": "attention_required" if blockers else "healthy",
            "generated_at": utc_now(),
            "steps": [step.to_dict() for step in steps],
            "blockers": [step.to_dict() for step in blockers],
            "flow": self._flow_summary(steps),
        }
    def _event_to_step(self, event: dict[str, Any]) -> GovernanceTimelineStep:
        payload = event.get("payload", {})
        kind = event.get("kind", "unknown")
        return GovernanceTimelineStep(
            id=event.get("id", f"event-{len(str(event))}"),
            kind=kind,
            title=self._title(kind, payload),
            actor=event.get("source", payload.get("actor", "unknown")),
            status=payload.get("status") or payload.get("decision") or event.get("status", "ok"),
            timestamp=event.get("created_at") or payload.get("created_at") or utc_now(),
            correlation_id=event.get("correlation_id") or payload.get("change_id") or payload.get("id"),
            detail=payload,
        )
    def _title(self, kind: str, payload: dict[str, Any]) -> str:
        titles = {
            "requirement_changed": "需求变化",
            "architecture_review_started": "架构审查开始",
            "architecture_review_completed": "架构审查完成",
            "test_gate_completed": "测试 Gate 完成",
            "user_acceptance_completed": "用户验收完成",
            "version_decision_created": "版本建议产生",
            "owner_gate_completed": "Owner Gate 完成",
            "promotion_completed": "Promotion 完成",
        }
        return payload.get("title") or payload.get("summary") or titles.get(kind, kind)
    def _flow_summary(self, steps: list[GovernanceTimelineStep]) -> dict[str, Any]:
        present = {step.kind for step in steps}
        required = [
            "requirement_changed",
            "architecture_review_completed",
            "test_gate_completed",
            "user_acceptance_completed",
            "version_decision_created",
        ]
        missing = [kind for kind in required if kind not in present]
        return {
            "required_steps": required,
            "missing_steps": missing,
            "ready_for_owner_gate": not missing,
        }
 __all__ = ["GovernanceTimelineStep", "GovernanceTimelineDashboard"]
@@ -1,59 +0,0 @@
 """Independent acceptance agents.
 Simulates independent roles: user representative and architecture reviewer.
 They provide acceptance evidence, not implementation decisions.
 """
 from dataclasses import dataclass, asdict
 from datetime import datetime, timezone
 from typing import Any
 def utc_now():
    return datetime.now(timezone.utc).isoformat()
@dataclass(frozen=True)
 class AcceptanceReport:
    agent: str
    role: str
    decision: str
    evidence: list[str]
    created_at: str = utc_now()
    def to_dict(self):
        return asdict(self)
 class UserAcceptanceAgent:
    def review(self, change: dict[str, Any]) -> AcceptanceReport:
        checks = [
            "user_requirement_match",
            "user_workflow_validation",
            "permission_visibility_check",
        ]
        return AcceptanceReport(
            "user-acceptance-agent",
            "user_representative",
            "pass" if change.get("user_facing") else "not_required",
            checks,
        )
 class ArchitectureAcceptanceAgent:
    def review(self, change: dict[str, Any]) -> AcceptanceReport:
        checks = [
            "architecture_consistency",
            "scalability_review",
            "security_boundary_review",
            "operability_review",
        ]
        return AcceptanceReport(
            "architecture-review-agent",
            "system_architect",
            "pass" if change.get("architecture_impact") else "not_required",
            checks,
        )
 __all__ = ["AcceptanceReport", "UserAcceptanceAgent", "ArchitectureAcceptanceAgent"]
@@ -1,92 +0,0 @@
 """Agent adapters for AIOS Search Gateway.
 All runtime agents should call SearchGatewayAgentAdapter instead of performing
 ad-hoc network search. This gives consistent provider fallback, profile
 resolution, evidence capture, audit and event visibility.
 """
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from typing import Any
 from .search_gateway_orchestrator import SearchGatewayOrchestrator
@dataclass(frozen=True)
 class AgentSearchContext:
    agent: str
    user_id: str | None = None
    group_id: str | None = None
    role: str = "user"
    group_profile: dict[str, Any] | None = None
    user_profile: dict[str, Any] | None = None
    session_profile: dict[str, Any] | None = None
    authorized_sources: list[str] = field(default_factory=list)
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
 class SearchGatewayAgentAdapter:
    def __init__(self, orchestrator: SearchGatewayOrchestrator | None = None):
        self.orchestrator = orchestrator or SearchGatewayOrchestrator()
    def search(
        self,
        query: str,
        context: AgentSearchContext,
        intent: str = "general",
        region: str | None = None,
        time_range: str | None = None,
        data_type: str | None = None,
        domain_profiles: list[str] | None = None,
        user_scope: str = "public",
        correlation_id: str | None = None,
    ) -> dict[str, Any]:
        payload = {
            "query": query,
            "requester": context.agent,
            "intent": intent,
            "region": region,
            "time_range": time_range,
            "data_type": data_type,
            "domain_profiles": domain_profiles or [],
            "user_scope": user_scope,
            "authorized_sources": context.authorized_sources,
            "group_profile": context.group_profile,
            "user_profile": context.user_profile,
            "session_profile": context.session_profile,
            "correlation_id": correlation_id,
        }
        return self.orchestrator.execute(payload)
 class OpenClawSearchClient(SearchGatewayAgentAdapter):
    def search_public(self, query: str, context: AgentSearchContext, **kwargs) -> dict[str, Any]:
        return self.search(query, context, user_scope="public", **kwargs)
 class HermesSearchClient(SearchGatewayAgentAdapter):
    def delegate_search(self, query: str, context: AgentSearchContext, **kwargs) -> dict[str, Any]:
        return self.search(query, context, **kwargs)
 class CodexSearchClient(SearchGatewayAgentAdapter):
    def technical_search(self, query: str, context: AgentSearchContext, **kwargs) -> dict[str, Any]:
        return self.search(query, context, intent=kwargs.pop("intent", "software_library"), **kwargs)
 class CustomGPTSearchClient(SearchGatewayAgentAdapter):
    def user_search(self, query: str, context: AgentSearchContext, **kwargs) -> dict[str, Any]:
        return self.search(query, context, **kwargs)
 __all__ = [
    "AgentSearchContext",
    "SearchGatewayAgentAdapter",
    "OpenClawSearchClient",
    "HermesSearchClient",
    "CodexSearchClient",
    "CustomGPTSearchClient",
 ]
@@ -1,73 +0,0 @@
 """Architect governance integration for Search Gateway.
 Search Gateway is a shared platform capability. The Architect Agent periodically
 reviews provider health, external search/browser capabilities, cost, security and
 privacy impact, then emits C/R/V recommendations.
 """
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from typing import Any
 def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()
@dataclass(frozen=True)
 class SearchArchitectureFinding:
    finding_id: str
    title: str
    impact: str
    recommendation: str
    evidence: list[str] = field(default_factory=list)
    created_at: str = field(default_factory=utc_now)
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
@dataclass(frozen=True)
 class SearchArchitectureDecision:
    status: str
    version_bump: str
    reason: str
    findings: list[SearchArchitectureFinding]
    owner_gate_required: bool
    created_at: str = field(default_factory=utc_now)
    def to_dict(self) -> dict[str, Any]:
        return {
            "status": self.status,
            "version_bump": self.version_bump,
            "reason": self.reason,
            "findings": [f.to_dict() for f in self.findings],
            "owner_gate_required": self.owner_gate_required,
            "created_at": self.created_at,
        }
 class SearchArchitectIntegration:
    def review(self, signals: dict[str, Any]) -> SearchArchitectureDecision:
        findings: list[SearchArchitectureFinding] = []
        if signals.get("provider_breaking_change"):
            findings.append(SearchArchitectureFinding("search-provider-breaking", "Provider breaking change", "breaking", "owner_review"))
        if signals.get("new_search_capability"):
            findings.append(SearchArchitectureFinding("search-new-capability", "New search capability available", "architecture", "evaluate_adoption"))
        if signals.get("cost_increase"):
            findings.append(SearchArchitectureFinding("search-cost", "Search provider cost increase", "minor", "optimize_provider_order"))
        if signals.get("security_advisory"):
            findings.append(SearchArchitectureFinding("search-security", "Search security advisory", "breaking", "owner_review"))
        if any(f.impact == "breaking" for f in findings):
            return SearchArchitectureDecision("action_required", "V", "breaking search ecosystem/security change", findings, True)
        if any(f.impact == "architecture" for f in findings):
            return SearchArchitectureDecision("optimization_recommended", "R", "new search capability may improve architecture", findings, True)
        if findings:
            return SearchArchitectureDecision("minor_change_recommended", "C", "compatible provider optimization", findings, False)
        return SearchArchitectureDecision("no_action", "none", "no relevant search architecture change", findings, False)
 __all__ = ["SearchArchitectureFinding", "SearchArchitectureDecision", "SearchArchitectIntegration"]
@@ -1,27 +0,0 @@
 """Audit storage for Search Gateway executions."""
 from __future__ import annotations
 import json
 from pathlib import Path
 from typing import Any
 class SearchAuditStore:
    def __init__(self, root: str = "runtime/search-gateway/audit"):
        self.root = Path(root)
        self.root.mkdir(parents=True, exist_ok=True)
    def append(self, record: dict[str, Any]) -> str:
        path = self.root / f"{record.get('request_id', 'search')}.json"
        path.write_text(json.dumps(record, ensure_ascii=False, indent=2), encoding="utf-8")
        return str(path)
    def list_recent(self) -> list[dict[str, Any]]:
        result = []
        for path in sorted(self.root.glob("*.json")):
            result.append(json.loads(path.read_text(encoding="utf-8")))
        return result[-50:]
 __all__ = ["SearchAuditStore"]
@@ -1,118 +0,0 @@
 """Search Gateway event bridge.
 Publishes normalized search lifecycle events into the AIOS Agent Event Bus when a
 bus is available, with an in-memory fallback for tests and offline runs.
 """
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from typing import Any, Protocol
 from uuid import uuid4
 def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()
 class SearchEventPublisher(Protocol):
    def publish(
        self,
        kind: str,
        source: str,
        target: str,
        summary: str,
        payload: dict[str, Any],
        correlation_id: str | None = None,
    ) -> dict[str, Any]:
        ...
@dataclass(frozen=True)
 class SearchEvent:
    id: str
    kind: str
    source: str
    target: str
    summary: str
    payload: dict[str, Any]
    correlation_id: str | None = None
    created_at: str = field(default_factory=utc_now)
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
 class InMemorySearchEventBus:
    def __init__(self):
        self.events: list[dict[str, Any]] = []
    def publish(
        self,
        kind: str,
        source: str,
        target: str,
        summary: str,
        payload: dict[str, Any],
        correlation_id: str | None = None,
    ) -> dict[str, Any]:
        event = SearchEvent(
            id=f"search_evt_{uuid4().hex[:12]}",
            kind=kind,
            source=source,
            target=target,
            summary=summary,
            payload=payload,
            correlation_id=correlation_id,
        ).to_dict()
        self.events.append(event)
        return event
 class SearchEventBridge:
    def __init__(self, bus: SearchEventPublisher | None = None):
        self.bus = bus or InMemorySearchEventBus()
    def search_started(self, request: dict[str, Any]) -> dict[str, Any]:
        return self.bus.publish(
            kind="search_started",
            source=request.get("requester", "unknown"),
            target="search-gateway",
            summary=request.get("query", "search started"),
            payload=request,
            correlation_id=request.get("correlation_id"),
        )
    def provider_attempted(self, request_id: str, attempt: dict[str, Any]) -> dict[str, Any]:
        return self.bus.publish(
            kind="search_provider_attempted",
            source="search-gateway",
            target=attempt.get("provider", "unknown"),
            summary=f"provider {attempt.get('provider', 'unknown')} {attempt.get('status', 'attempted')}",
            payload=attempt,
            correlation_id=request_id,
        )
    def fallback_triggered(self, request_id: str, reason: str) -> dict[str, Any]:
        return self.bus.publish(
            kind="search_fallback_triggered",
            source="search-gateway",
            target="search-provider-router",
            summary=reason,
            payload={"reason": reason},
            correlation_id=request_id,
        )
    def search_completed(self, response: dict[str, Any]) -> dict[str, Any]:
        return self.bus.publish(
            kind="search_completed",
            source="search-gateway",
            target=response.get("requester", "requester"),
            summary=f"search {response.get('status', 'completed')}",
            payload=response,
            correlation_id=response.get("request_id"),
        )
 __all__ = ["SearchEvent", "InMemorySearchEventBus", "SearchEventBridge"]
@@ -1,252 +0,0 @@
 """Governed general-purpose Search Gateway for AIOS.
 The gateway coordinates query planning, provider routing, fallback, evidence
 normalization and compliance checks. It is provider-agnostic: runtime workers plug
 in public or explicitly authorized search providers.
 Search is a shared capability. Groups and users may customize provider order,
 trusted domains, query expansion hints and domain profiles, but customization may
 not override hard authorization, privacy, safety or audit rules.
 """
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from typing import Any, Protocol
 from uuid import uuid4
 from .search_profile_resolver import DomainQueryProfileRegistry, SearchProfile, SearchProfileResolver
 def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()
@dataclass(frozen=True)
 class SearchRequest:
    query: str
    requester: str
    intent: str = "general"
    region: str | None = None
    time_range: str | None = None
    data_type: str | None = None
    user_scope: str = "public"
    authorized_sources: list[str] = field(default_factory=list)
    domain_profiles: list[str] = field(default_factory=list)
    group_profile: dict[str, Any] | None = None
    user_profile: dict[str, Any] | None = None
    session_profile: dict[str, Any] | None = None
    correlation_id: str | None = None
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
@dataclass(frozen=True)
 class SearchEvidence:
    title: str
    url: str | None
    source: str
    snippet: str
    evidence_type: str = "web"
    confidence: str = "medium"
    retrieved_at: str = field(default_factory=utc_now)
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
@dataclass(frozen=True)
 class SearchProviderAttempt:
    provider: str
    status: str
    result_count: int = 0
    reason: str = ""
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
@dataclass(frozen=True)
 class SearchResponse:
    request_id: str
    status: str
    answer: str
    evidence: list[SearchEvidence]
    attempts: list[SearchProviderAttempt]
    confidence: str
    expanded_queries: list[str] = field(default_factory=list)
    resolved_profile: dict[str, Any] | None = None
    need_clarification: list[str] = field(default_factory=list)
    blocked_reasons: list[str] = field(default_factory=list)
    created_at: str = field(default_factory=utc_now)
    def to_dict(self) -> dict[str, Any]:
        return {
            "request_id": self.request_id,
            "status": self.status,
            "answer": self.answer,
            "evidence": [e.to_dict() for e in self.evidence],
            "attempts": [a.to_dict() for a in self.attempts],
            "confidence": self.confidence,
            "expanded_queries": self.expanded_queries,
            "resolved_profile": self.resolved_profile,
            "need_clarification": self.need_clarification,
            "blocked_reasons": self.blocked_reasons,
            "created_at": self.created_at,
        }
 class SearchProvider(Protocol):
    name: str
    def search(self, request: SearchRequest, queries: list[str]) -> tuple[list[SearchEvidence], SearchProviderAttempt]:
        ...
 class QueryPlanner:
    def __init__(self, domain_registry: DomainQueryProfileRegistry | None = None):
        self.domain_registry = domain_registry or DomainQueryProfileRegistry()
    def expand(self, request: SearchRequest, profile: SearchProfile) -> list[str]:
        base = request.query.strip()
        queries = [base]
        if request.region:
            queries.append(f"{request.region} {base}")
        if request.time_range:
            queries.append(f"{base} {request.time_range}")
        domain_profiles = list(dict.fromkeys([request.intent] + request.domain_profiles + profile.domain_profiles))
        hints = list(profile.query_expansion_hints) + self.domain_registry.hints_for(domain_profiles)
        for hint in hints:
            queries.append(f"{base} {hint}")
        for domain in profile.trusted_domains:
            queries.append(f"site:{domain} {base}")
        return list(dict.fromkeys([q for q in queries if q]))
 class SearchPolicyGuard:
    """Blocks non-public or unauthorized collection patterns."""
    FORBIDDEN_TERMS = (
        "绕过登录",
        "绕过验证码",
        "验证码绕过",
        "破解登录",
        "个人成绩",
        "学生隐私",
        "受保护平台批量爬取",
    )
    def evaluate(self, request: SearchRequest, profile: SearchProfile) -> list[str]:
        q = request.query.lower()
        reasons: list[str] = []
        for term in self.FORBIDDEN_TERMS:
            if term.lower() in q:
                reasons.append(f"blocked_term:{term}")
        if request.user_scope not in {"public", "authorized_user_file", "authorized_link", "internal_archive"}:
            reasons.append(f"unsupported_scope:{request.user_scope}")
        if request.data_type in {"personal_grade", "student_private_data", "private_person_data"} and request.user_scope != "authorized_user_file":
            reasons.append("private_data_requires_user_authorized_file")
        if not profile.audit_required:
            reasons.append("audit_cannot_be_disabled")
        if not profile.fallback_required:
            reasons.append("fallback_cannot_be_disabled")
        return reasons
 class SearchGateway:
    def __init__(self, providers: list[SearchProvider] | None = None):
        self.providers = providers or []
        self.profile_resolver = SearchProfileResolver()
        self.planner = QueryPlanner()
        self.guard = SearchPolicyGuard()
    def search(self, request: SearchRequest) -> SearchResponse:
        request_id = request.correlation_id or f"search_{uuid4().hex[:12]}"
        profile = self.profile_resolver.resolve(request.group_profile, request.user_profile, request.session_profile)
        expanded_queries = self.planner.expand(request, profile)
        blocked = self.guard.evaluate(request, profile)
        if blocked:
            return SearchResponse(
                request_id=request_id,
                status="blocked",
                answer="该搜索请求涉及未授权访问、验证码/登录绕过、受保护平台批量抓取或个人隐私数据，已阻断。可改为搜索公开网页、官方公告、公开文档，或处理用户授权上传的文件/链接。",
                evidence=[],
                attempts=[],
                confidence="high",
                expanded_queries=expanded_queries,
                resolved_profile=profile.to_dict(),
                blocked_reasons=blocked,
            )
        evidence: list[SearchEvidence] = []
        attempts: list[SearchProviderAttempt] = []
        providers = self._ordered_providers(profile)
        for provider in providers:
            try:
                provider_evidence, attempt = provider.search(request, expanded_queries)
                evidence.extend(provider_evidence)
                attempts.append(attempt)
                if provider_evidence:
                    break
            except Exception as exc:
                attempts.append(SearchProviderAttempt(provider=getattr(provider, "name", "unknown"), status="failed", reason=str(exc)))
        if evidence:
            return SearchResponse(
                request_id=request_id,
                status="found",
                answer="已找到公开或授权来源结果。",
                evidence=evidence,
                attempts=attempts,
                confidence=self._confidence(evidence),
                expanded_queries=expanded_queries,
                resolved_profile=profile.to_dict(),
            )
        return SearchResponse(
            request_id=request_id,
            status="not_found",
            answer="未找到足够可靠的公开或授权结果；已记录 provider 尝试和失败原因。",
            evidence=[],
            attempts=attempts,
            confidence="low",
            expanded_queries=expanded_queries,
            resolved_profile=profile.to_dict(),
            need_clarification=self._clarification(request),
        )
    def _ordered_providers(self, profile: SearchProfile) -> list[SearchProvider]:
        if not self.providers:
            return []
        by_name = {provider.name: provider for provider in self.providers}
        ordered = [by_name[name] for name in profile.provider_order if name in by_name]
        remaining = [provider for provider in self.providers if provider.name not in {p.name for p in ordered}]
        return ordered + remaining
    def _confidence(self, evidence: list[SearchEvidence]) -> str:
        if any(e.confidence == "high" for e in evidence):
            return "high"
        if evidence:
            return "medium"
        return "low"
    def _clarification(self, request: SearchRequest) -> list[str]:
        if request.intent == "education_exam_data":
            return ["请补充学校名称、联考名称、年级、科目，以及需要试卷/答案还是成绩统计。"]
        if request.intent in {"company_research", "policy_regulation", "software_library", "product_research"}:
            return ["请补充主体名称、地区/版本、时间范围、需要的数据类型或可信来源范围。"]
        return ["请补充更具体的主体、平台、地区、时间、数据类型或授权文件/链接。"]
 __all__ = [
    "SearchRequest",
    "SearchEvidence",
    "SearchProviderAttempt",
    "SearchResponse",
    "SearchProvider",
    "QueryPlanner",
    "SearchPolicyGuard",
    "SearchGateway",
 ]
@@ -1,65 +0,0 @@
 """API endpoints for AIOS Search Gateway."""
 from typing import Any
 try:
    from fastapi import APIRouter
 except Exception:
    APIRouter = None
 from .search_gateway import SearchGateway, SearchRequest
 from .search_gateway_dashboard import SearchGatewayDashboard
 gateway = SearchGateway()
 dashboard = SearchGatewayDashboard()
 def search_gateway_request(payload: dict[str, Any]) -> dict[str, Any]:
    request = SearchRequest(**payload)
    return gateway.search(request).to_dict()
 def get_search_gateway_status() -> dict[str, Any]:
    return {
        "module": "search-gateway",
        "status": "ready",
        "providers_configured": len(gateway.providers),
        "policy_guard": "enabled",
        "profile_customization": "enabled",
        "fallback_required_when_provider_unavailable": True,
    }
 def get_search_gateway_overview() -> dict[str, Any]:
    return dashboard.overview()
 def build_router():
    if APIRouter is None:
        return None
    router = APIRouter(prefix="/admin/ai-evolution/search", tags=["search-gateway"])
    @router.get("/status", operation_id="getSearchGatewayStatus")
    def status():
        return get_search_gateway_status()
    @router.get("/overview", operation_id="getSearchGatewayOverview")
    def overview():
        return get_search_gateway_overview()
    @router.post("/query", operation_id="runSearchGatewayQuery")
    def query(payload: dict[str, Any]):
        return search_gateway_request(payload)
    return router
 router = build_router()
 __all__ = [
    "router",
    "search_gateway_request",
    "get_search_gateway_status",
    "get_search_gateway_overview",
 ]
@@ -1,85 +0,0 @@
 """Search Gateway Web UI dashboard model."""
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from typing import Any
 def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()
@dataclass(frozen=True)
 class SearchDashboardMetric:
    key: str
    title: str
    value: Any
    status: str = "ok"
    detail: str = ""
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
@dataclass(frozen=True)
 class SearchDashboardPanel:
    key: str
    title: str
    metrics: list[SearchDashboardMetric] = field(default_factory=list)
    def to_dict(self) -> dict[str, Any]:
        return {
            "key": self.key,
            "title": self.title,
            "metrics": [m.to_dict() for m in self.metrics],
        }
 class SearchGatewayDashboard:
    def overview(self, search_runs: list[dict[str, Any]] | None = None, providers: list[dict[str, Any]] | None = None) -> dict[str, Any]:
        search_runs = search_runs or []
        providers = providers or []
        blocked = [r for r in search_runs if r.get("status") == "blocked"]
        not_found = [r for r in search_runs if r.get("status") == "not_found"]
        provider_failures = [p for p in providers if p.get("status") == "failed"]
        panels = [
            SearchDashboardPanel(
                "search_health",
                "搜索健康度",
                [
                    SearchDashboardMetric("total_runs", "搜索请求数", len(search_runs)),
                    SearchDashboardMetric("not_found", "未找到结果", len(not_found), "warn" if not_found else "ok"),
                    SearchDashboardMetric("blocked", "合规阻断", len(blocked), "warn" if blocked else "ok"),
                ],
            ),
            SearchDashboardPanel(
                "provider_health",
                "Provider 可用性",
                [
                    SearchDashboardMetric("providers", "Provider 数量", len(providers)),
                    SearchDashboardMetric("provider_failures", "Provider 失败", len(provider_failures), "warn" if provider_failures else "ok"),
                ],
            ),
            SearchDashboardPanel(
                "evidence_quality",
                "证据质量",
                [
                    SearchDashboardMetric("with_evidence", "有证据的回答", len([r for r in search_runs if r.get("evidence")])) ,
                    SearchDashboardMetric("fallback_required", "要求多源降级", True),
                ],
            ),
        ]
        return {
            "module": "search-gateway",
            "route": "/admin/ai-evolution/search",
            "title": "AIOS 联网搜索治理中心",
            "status": "attention_required" if provider_failures else "healthy",
            "generated_at": utc_now(),
            "panels": [p.to_dict() for p in panels],
            "recent_runs": search_runs[-20:],
        }
 __all__ = ["SearchDashboardMetric", "SearchDashboardPanel", "SearchGatewayDashboard"]
@@ -1,41 +0,0 @@
 """Runtime orchestrator joining Search Gateway, registry, audit and events."""
 from __future__ import annotations
 from typing import Any
 from .search_audit_store import SearchAuditStore
 from .search_event_bridge import SearchEventBridge
 from .search_gateway import SearchGateway, SearchRequest
 from .search_provider_registry import SearchProviderRegistry
 class SearchGatewayOrchestrator:
    def __init__(self):
        self.registry = SearchProviderRegistry()
        self.audit = SearchAuditStore()
        self.events = SearchEventBridge()
        self.gateway = SearchGateway(self.registry.ordered())
    def execute(self, payload: dict[str, Any]) -> dict[str, Any]:
        request = SearchRequest(**payload)
        self.events.search_started(request.to_dict())
        response = self.gateway.search(request)
        result = response.to_dict()
        for attempt in result.get("attempts", []):
            self.events.provider_attempted(result["request_id"], attempt)
        if len(result.get("attempts", [])) > 1:
            self.events.fallback_triggered(result["request_id"], "multiple providers attempted")
        self.audit.append(result)
        self.events.search_completed(result)
        return result
    def status(self) -> dict[str, Any]:
        return {
            "registry": self.registry.status(),
            "audit": "enabled",
            "events": "enabled",
        }
 __all__ = ["SearchGatewayOrchestrator"]
@@ -1,124 +0,0 @@
 """Search profile resolver for general-purpose AIOS Search Gateway.
 The Search Gateway is a shared platform capability. Groups and users may customize
 provider preference, query expansion hints, trusted sources and domain profiles on
 top of the shared base profile. Customization is intentionally layered and cannot
 override hard safety, authorization or audit rules.
 """
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from typing import Any
 HARD_BLOCKED_BEHAVIORS = {
    "bypass_login",
    "bypass_captcha",
    "scrape_private_student_grades",
    "scrape_student_private_data",
    "bulk_scrape_protected_platforms",
    "collect_data_without_authorization",
 }
@dataclass(frozen=True)
 class SearchProfile:
    profile_id: str
    scope: str
    provider_order: list[str] = field(default_factory=lambda: [
        "cache",
        "official_site",
        "general_web",
        "vertical_source",
        "browser_fetch",
        "user_authorized_file",
        "internal_archive",
    ])
    trusted_domains: list[str] = field(default_factory=list)
    query_expansion_hints: list[str] = field(default_factory=list)
    domain_profiles: list[str] = field(default_factory=list)
    blocked_behaviors: set[str] = field(default_factory=lambda: set(HARD_BLOCKED_BEHAVIORS))
    audit_required: bool = True
    fallback_required: bool = True
    def to_dict(self) -> dict[str, Any]:
        data = asdict(self)
        data["blocked_behaviors"] = sorted(self.blocked_behaviors)
        return data
 class SearchProfileResolver:
    """Resolves base -> group -> user -> session search preferences."""
    def base_profile(self) -> SearchProfile:
        return SearchProfile(profile_id="base", scope="global")
    def resolve(
        self,
        group_profile: dict[str, Any] | None = None,
        user_profile: dict[str, Any] | None = None,
        session_profile: dict[str, Any] | None = None,
    ) -> SearchProfile:
        profile = self.base_profile().to_dict()
        for overlay in (group_profile, user_profile, session_profile):
            if overlay:
                profile = self._merge(profile, overlay)
        profile["blocked_behaviors"] = set(HARD_BLOCKED_BEHAVIORS)
        profile["audit_required"] = True
        profile["fallback_required"] = True
        return SearchProfile(**profile)
    def _merge(self, base: dict[str, Any], overlay: dict[str, Any]) -> dict[str, Any]:
        merged = dict(base)
        if overlay.get("provider_order"):
            merged["provider_order"] = list(overlay["provider_order"])
        for list_key in ("trusted_domains", "query_expansion_hints", "domain_profiles"):
            merged[list_key] = self._dedupe(list(merged.get(list_key, [])) + list(overlay.get(list_key, [])))
        merged["profile_id"] = overlay.get("profile_id", merged.get("profile_id", "custom"))
        merged["scope"] = overlay.get("scope", merged.get("scope", "custom"))
        return merged
    def _dedupe(self, items: list[str]) -> list[str]:
        return list(dict.fromkeys([item for item in items if item]))
 class DomainQueryProfileRegistry:
    """Optional domain-specific expansions for the general Search Gateway."""
    DEFAULT_PROFILES = {
        "education_exam_data": [
            "学校 年级 科目 试卷 答案 成绩统计",
            "联考 月考 期中 期末 质量检测",
        ],
        "policy_regulation": [
            "官方 通知 政策 解读 实施细则",
            "site:gov.cn OR site:edu.cn",
        ],
        "software_library": [
            "release notes changelog migration guide",
            "official documentation GitHub issues",
        ],
        "product_research": [
            "specs review comparison price availability",
            "official page manual support",
        ],
        "company_research": [
            "official announcement press release filings",
            "leadership product financial report",
        ],
    }
    def hints_for(self, domain_profiles: list[str]) -> list[str]:
        hints: list[str] = []
        for profile in domain_profiles:
            hints.extend(self.DEFAULT_PROFILES.get(profile, []))
        return list(dict.fromkeys(hints))
 __all__ = [
    "HARD_BLOCKED_BEHAVIORS",
    "SearchProfile",
    "SearchProfileResolver",
    "DomainQueryProfileRegistry",
 ]
@@ -1,121 +0,0 @@
 """Provider registry for the general AIOS Search Gateway.
 The registry makes provider availability explicit and auditable. Provider
 implementations are pluggable: production can register real public search,
 official-site, vertical-source, browser-fetch, authorized-file or archive
 providers without changing the Search Gateway core.
 """
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from typing import Any
 from .search_gateway import SearchEvidence, SearchProviderAttempt, SearchRequest
@dataclass(frozen=True)
 class SearchProviderConfig:
    name: str
    provider_type: str
    enabled: bool = True
    requires_secret: bool = False
    supports_public_web: bool = True
    supports_authorized_sources: bool = False
    health: str = "unknown"
    metadata: dict[str, Any] = field(default_factory=dict)
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
 class NullSearchProvider:
    """Safe placeholder provider used until runtime wires a real adapter."""
    def __init__(self, config: SearchProviderConfig):
        self.config = config
        self.name = config.name
    def search(self, request: SearchRequest, queries: list[str]):
        return [], SearchProviderAttempt(
            provider=self.name,
            status="not_configured" if self.config.enabled else "disabled",
            result_count=0,
            reason="provider adapter not wired in runtime",
        )
 class StaticArchiveProvider:
    """Small deterministic provider for tests and internal archived evidence."""
    name = "internal_archive"
    def __init__(self, records: list[dict[str, Any]] | None = None):
        self.records = records or []
    def search(self, request: SearchRequest, queries: list[str]):
        evidence: list[SearchEvidence] = []
        for record in self.records:
            haystack = " ".join([record.get("title", ""), record.get("snippet", ""), record.get("tags", "")]).lower()
            if any(q.lower() in haystack for q in queries):
                evidence.append(
                    SearchEvidence(
                        title=record.get("title", "archive result"),
                        url=record.get("url"),
                        source=self.name,
                        snippet=record.get("snippet", ""),
                        evidence_type="internal_archive",
                        confidence=record.get("confidence", "medium"),
                    )
                )
        return evidence, SearchProviderAttempt(provider=self.name, status="ok", result_count=len(evidence))
 class SearchProviderRegistry:
    DEFAULT_PROVIDER_CONFIGS = [
        SearchProviderConfig("cache", "cache", supports_public_web=False),
        SearchProviderConfig("official_site", "official_site"),
        SearchProviderConfig("general_web", "general_web"),
        SearchProviderConfig("vertical_source", "vertical_source"),
        SearchProviderConfig("browser_fetch", "browser_fetch"),
        SearchProviderConfig("user_authorized_file", "authorized_file", supports_public_web=False, supports_authorized_sources=True),
        SearchProviderConfig("internal_archive", "archive", supports_public_web=False),
    ]
    def __init__(self):
        self.configs: dict[str, SearchProviderConfig] = {}
        self.providers: dict[str, Any] = {}
        for config in self.DEFAULT_PROVIDER_CONFIGS:
            self.register_config(config)
    def register_config(self, config: SearchProviderConfig) -> None:
        self.configs[config.name] = config
        self.providers.setdefault(config.name, NullSearchProvider(config))
    def register_provider(self, name: str, provider: Any, config: SearchProviderConfig | None = None) -> None:
        if config:
            self.configs[name] = config
        elif name not in self.configs:
            self.configs[name] = SearchProviderConfig(name=name, provider_type="custom")
        self.providers[name] = provider
    def ordered(self, order: list[str] | None = None) -> list[Any]:
        order = order or list(self.configs)
        result = [self.providers[name] for name in order if name in self.providers and self.configs[name].enabled]
        result.extend([provider for name, provider in self.providers.items() if name not in order and self.configs[name].enabled])
        return result
    def status(self) -> dict[str, Any]:
        return {
            "providers": [self.configs[name].to_dict() for name in sorted(self.configs)],
            "enabled": len([c for c in self.configs.values() if c.enabled]),
            "configured": len([name for name, provider in self.providers.items() if not isinstance(provider, NullSearchProvider)]),
        }
 __all__ = [
    "SearchProviderConfig",
    "NullSearchProvider",
    "StaticArchiveProvider",
    "SearchProviderRegistry",
 ]
@@ -1,265 +0,0 @@
 <!DOCTYPE html>
 <html lang="en">
 <head>
 <meta charset="UTF-8">
 <meta name="viewport" content="width=device-width, initial-scale=1.0">
 <title>AI Evolution Dashboard</title>
 <style>
 * { margin: 0; padding: 0; box-sizing: border-box; }
 body { font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, Oxygen, Ubuntu, Cantarell, sans-serif; background: #0d1117; color: #c9d1d9; height: 100vh; overflow: hidden; }
 .app { display: flex; height: 100vh; }
 .sidebar { width: 260px; background: #161b22; border-right: 1px solid #30363d; padding: 20px; display: flex; flex-direction: column; gap: 16px; }
 .sidebar h1 { font-size: 18px; color: #f0f6fc; border-bottom: 1px solid #30363d; padding-bottom: 12px; margin-bottom: 8px; }
 .sidebar-nav { list-style: none; display: flex; flex-direction: column; gap: 4px; }
 .sidebar-nav li { padding: 8px 12px; border-radius: 6px; cursor: pointer; font-size: 14px; color: #8b949e; }
 .sidebar-nav li:hover, .sidebar-nav li.active { background: #1f6feb22; color: #58a6ff; }
 .sidebar-nav li i { margin-right: 8px; }
 .main { flex: 1; padding: 24px; overflow-y: auto; }
 .main-header { display: flex; justify-content: space-between; align-items: center; margin-bottom: 24px; }
 .main-header h2 { font-size: 20px; }
 .status-badge { padding: 4px 12px; border-radius: 12px; font-size: 12px; font-weight: 600; text-transform: uppercase; }
 .status-badge.ok { background: #1b3a2d; color: #3fb950; border: 1px solid #3fb95044; }
 .status-badge.degraded { background: #3a2e1b; color: #d29922; border: 1px solid #d2992244; }
 .status-badge.fail { background: #3a1b1b; color: #f85149; border: 1px solid #f8514944; }
 .card { background: #161b22; border: 1px solid #30363d; border-radius: 8px; padding: 16px; margin-bottom: 16px; }
 .card-title { font-size: 14px; color: #8b949e; margin-bottom: 8px; }
 .card-value { font-size: 28px; font-weight: 700; color: #f0f6fc; }
 .row { display: flex; gap: 16px; }
 .col { flex: 1; }
 .grid { display: grid; grid-template-columns: repeat(auto-fill, minmax(300px, 1fr)); gap: 16px; }
 pre { white-space: pre-wrap; word-break: break-word; }
 .command-item { padding: 12px; margin-bottom: 8px; border-radius: 6px; background: #0d1117; border: 1px solid #21262d; }
 .command-item .meta { display: flex; gap: 8px; align-items: center; font-size: 12px; color: #8b949e; }
 .command-item .name { font-size: 14px; color: #58a6ff; font-weight: 600; }
 .command-item .state { font-size: 11px; padding: 2px 8px; border-radius: 10px; text-transform: uppercase; }
 .state-active { background: #1b3a2d; color: #3fb950; }
 .state-running { background: #3a2e1b; color: #d29922; }
 .state-completed { background: #1b3a2d; color: #3fb950; }
 .state-failed { background: #3a1b1b; color: #f85149; }
 .loading { text-align: center; padding: 40px; color: #8b949e; }
 .error { padding: 20px; background: #3a1b1b; border: 1px solid #f8514944; border-radius: 8px; color: #f85149; margin-bottom: 16px; }
 .version-item { padding: 8px 12px; background: #0d1117; border: 1px solid #21262d; border-radius: 6px; margin-bottom: 4px; display: flex; justify-content: space-between; }
 .version-item .tag { color: #58a6ff; font-weight: 600; }
 .version-item .name { color: #8b949e; font-size: 13px; }
 .agent-item { background: #0d1117; border: 1px solid #21262d; border-radius: 6px; margin-bottom: 8px; padding: 12px; }
 .agent-item .agent-title { font-size: 14px; font-weight: 600; color: #c9d1d9; margin-bottom: 4px; }
 .agent-item .agent-meta { font-size: 12px; color: #8b949e; }
 .agent-run-item { padding: 8px 12px; margin-bottom: 4px; border-radius: 4px; background: #0d1117; border: 1px solid #21262d; display: flex; justify-content: space-between; align-items: center; }
 .agent-run-item .run-id { color: #58a6ff; font-size: 13px; font-weight: 600; }
 .agent-run-item .source { color: #8b949e; font-size: 12px; }
 </style>
 </head>
 <body>
 <div id="app" class="app">
  <div class="sidebar">
    <h1>🕹️ AI Evolution Dashboard</h1>
    <ul class="sidebar-nav">
      <li class="active" onclick="showPanel('overview')"><i>📊</i> Overview</li>
      <li onclick="showPanel('versions')"><i>🏷️</i> Versions</li>
      <li onclick="showPanel('commands')"><i>⚡</i> Commands</li>
      <li onclick="showPanel('agents')"><i>🤖</i> Agent Runs</li>
      <li onclick="showPanel('arch-owner')"><i>🔐</i> Owner Gate</li>
      <li onclick="showPanel('arch-approve')"><i>✓</i> Approve Changes</li>
    </ul>
  </div>
  <div class="main">
    <div id="panel-overview" class="panel">
      <div class="main-header">
        <h2>System Overview</h2>
        <div id="header-status" class="status-badge">Loading...</div>
      </div>
      <div id="loading" class="loading">Loading...</div>
      <div id="error" class="error" style="display:none;"></div>
      <div id="content">
        <div class="row">
          <div class="col"><div class="card"><div class="card-title">Current Baseline</div><div id="base" class="card-value">-</div></div></div>
          <div class="col"><div class="card"><div class="card-title">Git Branch</div><div id="branch" class="card-value">-</div></div></div>
          <div class="col"><div class="card"><div class="card-title">Uptime</div><div id="uptime" class="card-value">-</div></div></div>
        </div>
        <div class="card" style="margin-top:16px">
          <div class="card-title">Event Bus Status</div>
          <pre id="bus-status">-</pre>
        </div>
      </div>
    </div>
    <div id="panel-versions" class="panel" style="display:none;">
      <div class="main-header"><h2>Version Timeline</h2></div>
      <div id="versions-list">Loading...</div>
    </div>
    <div id="panel-commands" class="panel" style="display:none;">
      <div class="main-header"><h2>Command Lookup</h2></div>
      <div id="commands-list">Loading...</div>
    </div>
    <div id="panel-agents" class="panel" style="display:none;">
      <div class="main-header"><h2>Recent Agent Executions</h2></div>
    <div id="panel-arch-owner" class="panel" style="display:none;">
      <div class="main-header"><h2>🔐 Architecture Owner Gate</h2></div>
      <div id="arch-owner-content">Loading...</div>
    </div>
    <div id="panel-arch-approve" class="panel" style="display:none;">
      <div class="main-header"><h2>✓ Approve Architecture Changes</h2></div>
      <div id="arch-approve-content">Loading...</div>
    </div>
      <div id="agents-list">Loading...</div>
    </div>
  </div>
 </div>
 <script>
 async function fetchData() {
  const r = await fetch('/admin/ai-evolution/api/data');
  const data = await r.json();
  document.getElementById('loading').style.display = 'none';
  document.getElementById('error').style.display = 'none';
  document.getElementById('content').style.display = '';
  const hs = document.getElementById('header-status');
  hs.className = 'status-badge ' + (data.healthy ? 'ok' : 'fail');
  hs.textContent = data.healthy ? '▪ operational' : '⚠ degraded';
  document.getElementById('base').textContent = data.baseline || '-';
  document.getElementById('branch').textContent = data.main_branch || 'main';
  document.getElementById('uptime').textContent = data.uptime || '-';
  document.getElementById('bus-status').textContent = data.bus ? JSON.stringify(data.bus, null, 2) : 'no event bus data';
 }
 async function loadVersions() {
  const el = document.getElementById('versions-list');
  el.innerHTML = '<div class="loading">Loading versions...</div>';
  try {
    const r = await fetch('/admin/ai-evolution/api/versions');
    const data = await r.json();
    if (data.error) { el.innerHTML = '<div class="error">' + data.error + '</div>'; return; }
    let html = '<div class="grid">';
    for (const v of data.versions || []) {
      html += '<div class="card"><div class="card-title">' + v.tag + '</div><div class="card-value">' + (v.version || v.tag) + '</div></div>';
    }
    html += '</div>';
    el.innerHTML = html;
  } catch(e) {
    el.innerHTML = '<div class="error">Failed to load versions: ' + e.message + '</div>';
  }
 }
 async function loadCommands() {
  const el = document.getElementById('commands-list');
  el.innerHTML = '<div class="loading">Loading commands...</div>';
  try {
    const r = await fetch('/admin/ai-evolution/api/data');
    const data = await r.json();
    let cmds = data.commands || [];
    let html = '<div style="display:flex;gap:24px;margin-bottom:16px;">';
    html += '<div><div class="card"><div class="card-title">Total</div><div class="card-value">' + cmds.length + '</div></div></div>';
    html += '<div><div class="card"><div class="card-title">By Category</div>';
    const cats = {};
    for (const c of cmds) { const k = c.category || 'uncategorized'; cats[k] = (cats[k] || 0) + 1; }
    for (const [k, v] of Object.entries(cats)) html += '<div style="font-size:14px;color:#8b949e;margin:4px 0;">' + k + ': ' + v + '</div>';
    html += '</div></div></div>';
    for (const cmd of cmds.slice().reverse()) {
      html += '<div class="command-item"><div class="meta">' + cmd.category + ' | ' + cmd.filename + '</div><div class="name">' + cmd.title + '</div><div style="font-size:13px;color:#c9d1d9;margin-top:4px;">' + cmd.summary + '</div></div>';
    }
    el.innerHTML = html;
  } catch(e) {
    el.innerHTML = '<div class="error">Failed to load commands: ' + e.message + '</div>';
  }
 }
 async function loadAgents() {
  const el = document.getElementById('agents-list');
  el.innerHTML = '<div class="loading">Loading agent runs...</div>';
  try {
    const r = await fetch('/admin/ai-evolution/api/agents');
    const data = await r.json();
    if (data.error) { el.innerHTML = '<div class="error">' + data.error + '</div>'; return; }
    let html = '';
    for (const agent of data.agents || []) {
      html += '<div class="agent-item"><div class="agent-title">' + agent.name + ' <span class="state state-' + (agent.status === 'running' ? 'running' : (agent.status === 'ok' ? 'completed' : 'failed')) + '">' + agent.status + '</span></div><div class="agent-meta">' + (agent.runs || 0) + ' runs | last: ' + (agent.last_run || '-') + '</div></div>';
    }
    el.innerHTML = html || '<div style="padding:24px;color:#8b949e;">No agent runs recorded yet.</div>';
  } catch(e) {
    el.innerHTML = '<div class="error">Failed to load agents: ' + e.message + '</div>';
  }
 }
 function showPanel(name) {
 async function loadOwnerGate() {
  const el = document.getElementById('arch-owner-content');
  el.innerHTML = '<div class="loading">Loading owner gate data...</div>';
  try {
    const r = await fetch('/admin/ai-evolution/api/owner-gate');
    const data = await r.json();
    let html = '<div class="row">';
    html += '<div class="col"><div class="card"><div class="card-title">Pending Requests</div><div class="card-value">' + data.pending_count + '</div></div></div>';
    html += '<div class="col"><div class="card"><div class="card-title">Approved</div><div class="card-value">' + data.approved_count + '</div></div></div>';
    html += '<div class="col"><div class="card"><div class="card-title">Rejected</div><div class="card-value">' + data.rejected_count + '</div></div></div>';
    html += '</div>';
    html += '<div class="card"><div class="card-title">Pending Approval Requests</div>';
    if (data.pending_requests && data.pending_requests.length) {
      for (const req of data.pending_requests) {
        html += '<div class="command-item">';
        html += '<div class="name">' + (req.title || req.change_id) + '</div>';
        html += '<div class="meta">ID: ' + req.change_id + ' | Risk: ' + req.risk + '</div>';
        html += '<div style="font-size:13px;color:#c9d1d9;margin-top:4px;">' + (req.reason || '') + '</div>';
        html += '</div>';
      }
    } else {
      html += '<div style="color:#8b949e;padding:12px;">No pending requests.</div>';
    }
    html += '</div>';
    el.innerHTML = html;
  } catch(e) {
    el.innerHTML = '<div class="error">Failed to load: ' + e.message + '</div>';
  }
 }
 async function loadApprove() {
  const el = document.getElementById('arch-approve-content');
  el.innerHTML = '<div class="loading">Loading pending approvals...</div>';
  try {
    const r = await fetch('/admin/ai-evolution/api/owner-gate');
    const data = await r.json();
    let html = '';
    if (data.pending_requests && data.pending_requests.length) {
      html += '<div class="card"><div class="card-title">Pending changes requiring your approval</div>';
      for (const req of data.pending_requests) {
        html += '<div class="command-item">';
        html += '<div class="name">' + (req.title || req.change_id) + '</div>';
        html += '<div class="meta">Risk: <strong>' + req.risk + '</strong></div>';
        html += '<div style="font-size:13px;color:#c9d1d9;margin:8px 0;">' + (req.reason || '') + '</div>';
        html += '<div style="display:flex;gap:8px;">';
        html += '<button onclick="ownerAction.classList.add('\'approved\');this.disabled=true" style="padding:6px 16px;background:#238636;color:#fff;border:none;border-radius:6px;cursor:pointer;font-size:13px;">✓ Approve</button>';
        html += '</div></div>';
      }
      html += '</div>';
    } else {
      html += '<div class="card"><div style="padding:24px;text-align:center;color:#8b949e;">✅ All changes processed. No pending approval requests.</div></div>';
    }
    el.innerHTML = html;
  } catch(e) {
    el.innerHTML = '<div class="error">Failed to load: ' + e.message + '</div>';
  }
 }
  document.querySelectorAll('.panel').forEach(p => p.style.display = 'none');
  document.getElementById('panel-' + name).style.display = '';
  document.querySelectorAll('.sidebar-nav li').forEach(li => li.classList.remove('active'));
  document.querySelector('.sidebar-nav li:nth-child(' + ({overview:1,versions:2,commands:3,agents:4}[name] || 1) + ')').classList.add('active');
  if (name === 'overview') {
    document.getElementById('panel-overview').innerHTML = '<div class="main-header"><h2>System Overview</h2><div id="header-status" class="status-badge">Loading...</div></div><div id="loading" class="loading">Loading...</div><div id="error" class="error" style="display:none;"></div><div id="content"><div class="row"><div class="col"><div class="card"><div class="card-title">Current Baseline</div><div id="base" class="card-value">-</div></div></div><div class="col"><div class="card"><div class="card-title">Git Branch</div><div id="branch" class="card-value">-</div></div></div><div class="col"><div class="card"><div class="card-title">Uptime</div><div id="uptime" class="card-value">-</div></div></div></div><div class="card" style="margin-top:16px"><div class="card-title">Event Bus Status</div><pre id="bus-status">-</pre></div></div>';
    fetchData();
  }
  const loaders = {
    'arch-owner': loadOwnerGate,
    'arch-approve': loadApprove,
    versions: loadVersions,
    commands: loadCommands,
    agents: loadAgents,
  };
  const fn = loaders[name];
  if (fn) fn();
 }
 fetchData();
 </script>
 </body>
 </html>
@@ -1,41 +0,0 @@
 """Task Center action API model for Web UI buttons."""
 from typing import Any
 try:
    from fastapi import APIRouter
 except Exception:
    APIRouter = None
 from .task_lifecycle import TaskLifecycle
 lifecycle = TaskLifecycle()
 def get_task_actions(task: dict[str, Any], role: str) -> dict[str, Any]:
    return {
        "task_id": task.get("task_id"),
        "actions": lifecycle.available_actions(task, role),
    }
 def execute_task_action(task: dict[str, Any], action: str, role: str, reason: str | None = None):
    return lifecycle.transition(task, action, role, reason)
 def build_router():
    if APIRouter is None:
        return None
    router = APIRouter(prefix="/admin/ai-evolution/tasks", tags=["task-actions"])
    @router.post("/{task_id}/actions/{action}", operation_id="executeTaskAction")
    def action(task_id: str, action: str, payload: dict[str, Any]):
        return execute_task_action(payload.get("task", {}), action, payload.get("role", "user"), payload.get("reason"))
    return router
 router = build_router()
 __all__ = ["router", "get_task_actions", "execute_task_action"]
@@ -1,115 +0,0 @@
 """AIOS Task Center domain model.
 The Task Center gives owners and authorized users a single Web UI page showing:
 who assigned a task, who is executing it, progress, elapsed time, remaining
 work, and whether completion notifications were sent.
 """
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from typing import Any
 def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()
 def parse_time(value: str | None) -> datetime | None:
    if not value:
        return None
    return datetime.fromisoformat(value.replace("Z", "+00:00"))
@dataclass(frozen=True)
 class TaskNotificationStatus:
    required: bool = True
    sent: bool = False
    channel: str | None = None
    sent_at: str | None = None
    recipient_scope: str | None = None
    last_error: str | None = None
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
@dataclass(frozen=True)
 class TaskProgress:
    percent: int = 0
    current_step: str = "pending"
    completed_steps: list[str] = field(default_factory=list)
    remaining_steps: list[str] = field(default_factory=list)
    blocked_reason: str | None = None
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
@dataclass(frozen=True)
 class AIOSTaskRecord:
    task_id: str
    title: str
    assigned_by: str
    executor: str
    source: str
    status: str = "pending"
    priority: str = "normal"
    created_at: str = field(default_factory=utc_now)
    started_at: str | None = None
    updated_at: str = field(default_factory=utc_now)
    completed_at: str | None = None
    progress: TaskProgress = field(default_factory=TaskProgress)
    notification: TaskNotificationStatus = field(default_factory=TaskNotificationStatus)
    owner_gate_required: bool = False
    owner_gate_status: str | None = None
    evidence: list[str] = field(default_factory=list)
    def elapsed_seconds(self) -> int:
        start = parse_time(self.started_at) or parse_time(self.created_at)
        end = parse_time(self.completed_at) or datetime.now(timezone.utc)
        return int((end - start).total_seconds()) if start else 0
    def to_dict(self) -> dict[str, Any]:
        data = asdict(self)
        data["elapsed_seconds"] = self.elapsed_seconds()
        data["progress"] = self.progress.to_dict()
        data["notification"] = self.notification.to_dict()
        return data
 class TaskCenter:
    def summarize(self, tasks: list[AIOSTaskRecord] | list[dict[str, Any]] | None = None) -> dict[str, Any]:
        raw_tasks = tasks or []
        task_dicts = [t.to_dict() if hasattr(t, "to_dict") else dict(t) for t in raw_tasks]
        active = [t for t in task_dicts if t.get("status") in {"running", "pending", "blocked"}]
        blocked = [t for t in task_dicts if t.get("status") == "blocked"]
        completed = [t for t in task_dicts if t.get("status") == "completed"]
        notification_pending = [
            t for t in completed
            if t.get("notification", {}).get("required") and not t.get("notification", {}).get("sent")
        ]
        return {
            "module": "task-center",
            "route": "/admin/ai-evolution/tasks",
            "title": "AIOS 任务中心",
            "status": "attention_required" if blocked or notification_pending else "healthy",
            "generated_at": utc_now(),
            "summary": {
                "total": len(task_dicts),
                "active": len(active),
                "blocked": len(blocked),
                "completed": len(completed),
                "notification_pending": len(notification_pending),
            },
            "tasks": sorted(task_dicts, key=lambda t: t.get("updated_at", ""), reverse=True),
        }
 __all__ = [
    "TaskNotificationStatus",
    "TaskProgress",
    "AIOSTaskRecord",
    "TaskCenter",
 ]
@@ -1,34 +0,0 @@
 """Task Center API."""
 from typing import Any
 try:
    from fastapi import APIRouter
 except Exception:
    APIRouter = None
 from .task_center import TaskCenter
 task_center = TaskCenter()
 def get_task_overview(tasks: list[dict[str, Any]] | None = None):
    return task_center.summarize(tasks)
 def build_router():
    if APIRouter is None:
        return None
    router = APIRouter(prefix="/admin/ai-evolution/tasks", tags=["task-center"])
    @router.get("/overview", operation_id="getTaskCenterOverview")
    def overview():
        return get_task_overview()
    return router
 router = build_router()
 __all__ = ["router", "get_task_overview"]
@@ -1,81 +0,0 @@
 """Task Center visualization model."""
 from __future__ import annotations
 from typing import Any
 class TaskDashboardView:
    def build(self, tasks: list[dict[str, Any]] | None = None) -> dict[str, Any]:
        tasks = tasks or []
        return {
            "module": "task-center",
            "route": "/admin/ai-evolution/tasks",
            "title": "AIOS Task Center",
            "views": {
                "kanban": self._kanban(tasks),
                "timeline": self._timeline(tasks),
                "flow": self._flow(tasks),
                "statistics": self._statistics(tasks),
            },
            "tasks": [self._task_card(task) for task in tasks],
        }
    def _task_card(self, task: dict[str, Any]) -> dict[str, Any]:
        return {
            "task_id": task.get("task_id"),
            "title": task.get("title"),
            "assigned_by": task.get("assigned_by"),
            "executor": task.get("executor"),
            "status": task.get("status"),
            "progress": task.get("progress", {}),
            "elapsed_seconds": task.get("elapsed_seconds", 0),
            "remaining_steps": task.get("progress", {}).get("remaining_steps", []),
            "notification": task.get("notification", {}),
            "actions": [
                "start",
                "update_progress",
                "block",
                "request_approval",
                "approve",
                "reject",
                "retry",
                "complete",
                "notify",
                "close",
            ],
        }
    def _kanban(self, tasks):
        columns = {"pending": [], "running": [], "blocked": [], "waiting_approval": [], "completed": []}
        for task in tasks:
            columns.setdefault(task.get("status", "pending"), []).append(task.get("task_id"))
        return columns
    def _timeline(self, tasks):
        return [
            {
                "task_id": task.get("task_id"),
                "updated_at": task.get("updated_at"),
                "status": task.get("status"),
            }
            for task in tasks
        ]
    def _flow(self, tasks):
        return {
            "nodes": ["created", "assigned", "running", "blocked", "approval", "completed", "closed"],
            "task_count": len(tasks),
        }
    def _statistics(self, tasks):
        return {
            "total": len(tasks),
            "running": len([t for t in tasks if t.get("status") == "running"]),
            "blocked": len([t for t in tasks if t.get("status") == "blocked"]),
            "waiting_approval": len([t for t in tasks if t.get("status") == "waiting_approval"]),
            "completed": len([t for t in tasks if t.get("status") == "completed"]),
        }
 __all__ = ["TaskDashboardView"]
@@ -1,129 +0,0 @@
 """Task Center event bridge.
 Converts task lifecycle actions and agent execution events into normalized AIOS
 Task Center events. Uses duck-typed Agent Event Bus publisher when available and
 an in-memory fallback for tests/offline runs.
 """
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from typing import Any, Protocol
 from uuid import uuid4
 def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()
 class TaskEventPublisher(Protocol):
    def publish(
        self,
        kind: str,
        source: str,
        target: str,
        summary: str,
        payload: dict[str, Any],
        correlation_id: str | None = None,
    ) -> dict[str, Any]:
        ...
@dataclass(frozen=True)
 class TaskEvent:
    id: str
    kind: str
    source: str
    target: str
    summary: str
    payload: dict[str, Any]
    correlation_id: str | None = None
    created_at: str = field(default_factory=utc_now)
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
 class InMemoryTaskEventBus:
    def __init__(self):
        self.events: list[dict[str, Any]] = []
    def publish(
        self,
        kind: str,
        source: str,
        target: str,
        summary: str,
        payload: dict[str, Any],
        correlation_id: str | None = None,
    ) -> dict[str, Any]:
        event = TaskEvent(
            id=f"task_evt_{uuid4().hex[:12]}",
            kind=kind,
            source=source,
            target=target,
            summary=summary,
            payload=payload,
            correlation_id=correlation_id,
        ).to_dict()
        self.events.append(event)
        return event
 class TaskEventBridge:
    def __init__(self, bus: TaskEventPublisher | None = None):
        self.bus = bus or InMemoryTaskEventBus()
    def task_created(self, task: dict[str, Any]) -> dict[str, Any]:
        return self.bus.publish(
            kind="task_created",
            source=task.get("assigned_by", "unknown"),
            target=task.get("executor", "task-center"),
            summary=task.get("title", "task created"),
            payload=task,
            correlation_id=task.get("task_id"),
        )
    def task_updated(self, task: dict[str, Any], action: str) -> dict[str, Any]:
        return self.bus.publish(
            kind="task_updated",
            source=task.get("executor", "task-center"),
            target="task-center",
            summary=f"task {action}: {task.get('status')}",
            payload={"action": action, "task": task},
            correlation_id=task.get("task_id"),
        )
    def task_blocked(self, task: dict[str, Any], reason: str) -> dict[str, Any]:
        return self.bus.publish(
            kind="task_blocked",
            source=task.get("executor", "unknown"),
            target="owner-gate",
            summary=reason,
            payload={"task": task, "reason": reason},
            correlation_id=task.get("task_id"),
        )
    def approval_requested(self, task: dict[str, Any], reason: str) -> dict[str, Any]:
        return self.bus.publish(
            kind="task_approval_requested",
            source=task.get("executor", "unknown"),
            target="owner-gate",
            summary=reason,
            payload={"task": task, "reason": reason},
            correlation_id=task.get("task_id"),
        )
    def notification_sent(self, task: dict[str, Any], channel: str) -> dict[str, Any]:
        return self.bus.publish(
            kind="task_notification_sent",
            source="notification-agent",
            target=task.get("assigned_by", "owner"),
            summary=f"notification sent via {channel}",
            payload={"task": task, "channel": channel},
            correlation_id=task.get("task_id"),
        )
 __all__ = ["TaskEvent", "InMemoryTaskEventBus", "TaskEventBridge"]
@@ -1,147 +0,0 @@
 """AIOS Task Lifecycle.
 Defines a governed task state machine for task creation, assignment, execution,
 blocking, approval, retry, completion, notification and closure.
 """
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from typing import Any
 def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()
 TERMINAL_STATES = {"completed", "cancelled", "failed", "closed"}
@dataclass(frozen=True)
 class TaskTransition:
    action: str
    from_states: set[str]
    to_state: str
    required_role: str = "agent"
    requires_reason: bool = False
    notify: bool = False
    def to_dict(self) -> dict[str, Any]:
        data = asdict(self)
        data["from_states"] = sorted(self.from_states)
        return data
 class TaskLifecycle:
    """Central task state machine."""
    TRANSITIONS = {
        "assign": TaskTransition("assign", {"new", "pending"}, "assigned", "dispatcher"),
        "start": TaskTransition("start", {"assigned", "pending"}, "running", "executor"),
        "update_progress": TaskTransition("update_progress", {"running"}, "running", "executor"),
        "block": TaskTransition("block", {"running", "assigned", "pending"}, "blocked", "executor", requires_reason=True, notify=True),
        "request_approval": TaskTransition("request_approval", {"running", "blocked"}, "waiting_approval", "executor", requires_reason=True, notify=True),
        "approve": TaskTransition("approve", {"waiting_approval"}, "approved", "owner", notify=True),
        "reject": TaskTransition("reject", {"waiting_approval"}, "rejected", "owner", requires_reason=True, notify=True),
        "resume": TaskTransition("resume", {"blocked", "approved"}, "running", "executor"),
        "retry": TaskTransition("retry", {"failed", "blocked"}, "running", "executor", notify=True),
        "complete": TaskTransition("complete", {"running", "approved"}, "completed", "executor", notify=True),
        "notify": TaskTransition("notify", {"completed"}, "completed", "notifier"),
        "close": TaskTransition("close", {"completed", "cancelled", "failed"}, "closed", "owner"),
        "cancel": TaskTransition("cancel", {"new", "pending", "assigned", "running", "blocked", "waiting_approval"}, "cancelled", "owner", requires_reason=True, notify=True),
        "fail": TaskTransition("fail", {"running", "assigned"}, "failed", "executor", requires_reason=True, notify=True),
    }
    ROLE_ORDER = {
        "user": 1,
        "agent": 2,
        "executor": 3,
        "dispatcher": 4,
        "notifier": 4,
        "group_admin": 5,
        "owner": 9,
    }
    def available_actions(self, task: dict[str, Any], role: str) -> list[dict[str, Any]]:
        state = task.get("status", "new")
        actions = []
        for name, transition in self.TRANSITIONS.items():
            allowed_state = state in transition.from_states
            allowed_role = self._role_allows(role, transition.required_role)
            actions.append({
                "action": name,
                "label": self._label(name),
                "enabled": allowed_state and allowed_role,
                "from_state": state,
                "to_state": transition.to_state,
                "required_role": transition.required_role,
                "requires_reason": transition.requires_reason,
                "notify": transition.notify,
                "disabled_reason": None if allowed_state and allowed_role else self._disabled_reason(allowed_state, allowed_role, transition),
            })
        return actions
    def transition(self, task: dict[str, Any], action: str, actor_role: str, reason: str | None = None) -> dict[str, Any]:
        if action not in self.TRANSITIONS:
            raise ValueError(f"unknown task action: {action}")
        transition = self.TRANSITIONS[action]
        state = task.get("status", "new")
        if state not in transition.from_states:
            raise ValueError(f"action {action} not allowed from state {state}")
        if not self._role_allows(actor_role, transition.required_role):
            raise PermissionError(f"role {actor_role} cannot execute {action}; requires {transition.required_role}")
        if transition.requires_reason and not reason:
            raise ValueError(f"action {action} requires reason")
        updated = dict(task)
        updated["status"] = transition.to_state
        updated["updated_at"] = utc_now()
        history = list(updated.get("history", []))
        history.append({
            "action": action,
            "from": state,
            "to": transition.to_state,
            "actor_role": actor_role,
            "reason": reason,
            "ts": utc_now(),
        })
        updated["history"] = history
        if action == "complete":
            updated["completed_at"] = utc_now()
            notification = dict(updated.get("notification", {}))
            notification.setdefault("required", True)
            notification.setdefault("sent", False)
            updated["notification"] = notification
        return updated
    def _role_allows(self, role: str, required: str) -> bool:
        return self.ROLE_ORDER.get(role, 0) >= self.ROLE_ORDER.get(required, 0)
    def _label(self, action: str) -> str:
        labels = {
            "assign": "分配任务",
            "start": "开始执行",
            "update_progress": "更新进度",
            "block": "标记阻塞",
            "request_approval": "请求审批",
            "approve": "批准",
            "reject": "拒绝",
            "resume": "继续执行",
            "retry": "重试",
            "complete": "标记完成",
            "notify": "发送通知",
            "close": "关闭任务",
            "cancel": "取消任务",
            "fail": "标记失败",
        }
        return labels.get(action, action)
    def _disabled_reason(self, allowed_state: bool, allowed_role: bool, transition: TaskTransition) -> str:
        if not allowed_state:
            return "当前状态不允许该操作"
        if not allowed_role:
            return f"需要角色: {transition.required_role}"
        return "不可用"
 __all__ = ["TaskTransition", "TaskLifecycle", "TERMINAL_STATES"]
@@ -1,50 +0,0 @@
 """Task Center orchestrator.
 Coordinates task ingestion, persistence, lifecycle updates, events and statistics.
 """
 from __future__ import annotations
 from typing import Any
 from .task_event_bridge import TaskEventBridge
 from .task_source_ingestor import TaskSourceIngestor
 from .task_statistics import TaskStatistics
 from .task_store import TaskStore
 class TaskOrchestrator:
    def __init__(self):
        self.ingestor = TaskSourceIngestor()
        self.store = TaskStore()
        self.events = TaskEventBridge()
        self.statistics = TaskStatistics()
    def create_from_command(self, path: str, bucket: str = "active") -> dict[str, Any]:
        task = self.ingestor.command_file_to_task(path, bucket)
        self.store.append(task)
        self.events.task_created(task)
        return task
    def create_from_agent_execution(self, record: dict[str, Any]) -> dict[str, Any]:
        task = self.ingestor.agent_execution_to_task(record)
        self.store.append(task)
        self.events.task_created(task)
        return task
    def sync_event(self, event: dict[str, Any]) -> dict[str, Any] | None:
        task = self.ingestor.event_to_task(event)
        if task:
            self.store.append(task)
            self.events.task_updated(task, event.get("kind", "event"))
        return task
    def dashboard(self) -> dict[str, Any]:
        tasks = self.store.latest()
        return {
            "tasks": tasks,
            "statistics": self.statistics.summarize(tasks),
        }
 __all__ = ["TaskOrchestrator"]
@@ -1,159 +0,0 @@
 """Task source ingestion for AIOS Task Center.
 Converts Command Watcher files, OpenClaw/Hermes/Codex execution records and
 Generic Agent Event Bus records into normalized Task Center task dictionaries.
 """
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any
 def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()
@dataclass(frozen=True)
 class TaskSourceRecord:
    source_type: str
    source_id: str
    title: str
    assigned_by: str
    executor: str
    status: str
    priority: str = "normal"
    created_at: str = field(default_factory=utc_now)
    updated_at: str = field(default_factory=utc_now)
    evidence: list[str] = field(default_factory=list)
    metadata: dict[str, Any] = field(default_factory=dict)
    def to_task(self) -> dict[str, Any]:
        return {
            "task_id": f"task-{self.source_type}-{self.source_id}",
            "title": self.title,
            "assigned_by": self.assigned_by,
            "executor": self.executor,
            "source": self.source_type,
            "status": self.status,
            "priority": self.priority,
            "created_at": self.created_at,
            "updated_at": self.updated_at,
            "progress": {
                "percent": self._progress_percent(),
                "current_step": self.status,
                "completed_steps": [],
                "remaining_steps": [],
                "blocked_reason": self.metadata.get("blocked_reason"),
            },
            "notification": {
                "required": True,
                "sent": bool(self.metadata.get("notification_sent")),
                "channel": self.metadata.get("notification_channel"),
                "sent_at": self.metadata.get("notification_sent_at"),
            },
            "evidence": self.evidence,
            "metadata": self.metadata,
        }
    def _progress_percent(self) -> int:
        mapping = {
            "new": 0,
            "pending": 5,
            "assigned": 15,
            "running": 50,
            "blocked": 50,
            "waiting_approval": 60,
            "approved": 70,
            "completed": 100,
            "closed": 100,
            "failed": 100,
            "cancelled": 100,
        }
        return mapping.get(self.status, 0)
 class TaskSourceIngestor:
    COMMAND_STATUS_MAP = {
        "active": "pending",
        "running": "running",
        "completed": "completed",
        "failed": "failed",
        "archive": "closed",
    }
    def command_file_to_task(self, path: str, bucket: str = "active") -> dict[str, Any]:
        p = Path(path)
        title = p.stem
        return TaskSourceRecord(
            source_type="command_watcher",
            source_id=title,
            title=title,
            assigned_by="command-watcher",
            executor=self._infer_executor(title),
            status=self.COMMAND_STATUS_MAP.get(bucket, "pending"),
            priority="high" if "P0" in title or "fix" in title.lower() else "normal",
            evidence=[str(p)],
            metadata={"bucket": bucket},
        ).to_task()
    def agent_execution_to_task(self, record: dict[str, Any]) -> dict[str, Any]:
        source_id = record.get("execution_id", record.get("id", "unknown"))
        status = record.get("status", "running")
        return TaskSourceRecord(
            source_type="agent_execution",
            source_id=source_id,
            title=record.get("title", f"Agent execution {source_id}"),
            assigned_by=record.get("assigned_by", record.get("entry_agent", "agent-system")),
            executor=record.get("closed_by", record.get("entry_agent", "unknown-agent")),
            status=status,
            priority=record.get("priority", "normal"),
            created_at=record.get("started_at", utc_now()),
            updated_at=record.get("ended_at", utc_now()),
            evidence=[record.get("execution_id", source_id)],
            metadata=record,
        ).to_task()
    def event_to_task(self, event: dict[str, Any]) -> dict[str, Any] | None:
        kind = event.get("kind")
        payload = event.get("payload", {})
        if kind not in {"task_created", "task_updated", "agent_started", "agent_completed", "agent_failed", "search_started", "search_completed"}:
            return None
        source_id = event.get("correlation_id") or event.get("id", "event")
        status = self._status_from_event(kind, payload)
        return TaskSourceRecord(
            source_type="event_bus",
            source_id=source_id,
            title=payload.get("title") or event.get("summary", kind),
            assigned_by=event.get("source", "event-bus"),
            executor=event.get("target", "task-center"),
            status=status,
            created_at=event.get("created_at", utc_now()),
            updated_at=event.get("created_at", utc_now()),
            evidence=[event.get("id", source_id)],
            metadata={"event_kind": kind, "payload": payload},
        ).to_task()
    def _status_from_event(self, kind: str, payload: dict[str, Any]) -> str:
        if kind.endswith("completed"):
            return "completed"
        if kind.endswith("failed"):
            return "failed"
        return payload.get("status", "running")
    def _infer_executor(self, title: str) -> str:
        lower = title.lower()
        if "openclaw" in lower:
            return "openclaw"
        if "hermes" in lower:
            return "hermes"
        if "codex" in lower:
            return "codex"
        if "search" in lower:
            return "openclaw"
        return "openclaw"
 __all__ = ["TaskSourceRecord", "TaskSourceIngestor"]
@@ -1,33 +0,0 @@
 """Task Center statistics."""
 from typing import Any
 class TaskStatistics:
    def summarize(self, tasks: list[dict[str, Any]] | None = None) -> dict[str, Any]:
        tasks = tasks or []
        completed = [t for t in tasks if t.get("status") == "completed"]
        blocked = [t for t in tasks if t.get("status") == "blocked"]
        waiting = [t for t in tasks if t.get("status") == "waiting_approval"]
        elapsed = [t.get("elapsed_seconds", 0) for t in tasks if t.get("elapsed_seconds") is not None]
        return {
            "total": len(tasks),
            "completed": len(completed),
            "blocked": len(blocked),
            "waiting_approval": len(waiting),
            "completion_rate": round(len(completed) / len(tasks) * 100, 2) if tasks else 0,
            "average_elapsed_seconds": round(sum(elapsed) / len(elapsed), 2) if elapsed else 0,
            "top_blockers": self._top_values(blocked, "blocker"),
            "executor_distribution": self._top_values(tasks, "executor"),
        }
    def _top_values(self, tasks: list[dict[str, Any]], key: str):
        result: dict[str, int] = {}
        for task in tasks:
            value = task.get(key)
            if value:
                result[value] = result.get(value, 0) + 1
        return result
 __all__ = ["TaskStatistics"]
@@ -1,29 +0,0 @@
 """Persistent Task Center store."""
 from __future__ import annotations
 import json
 from pathlib import Path
 from typing import Any
 class TaskStore:
    def __init__(self, root: str = "runtime/task-center"):
        self.root = Path(root)
        self.root.mkdir(parents=True, exist_ok=True)
        self.path = self.root / "tasks.jsonl"
    def append(self, task: dict[str, Any]) -> None:
        with self.path.open("a", encoding="utf-8") as f:
            f.write(json.dumps(task, ensure_ascii=False) + "\n")
    def list_all(self) -> list[dict[str, Any]]:
        if not self.path.exists():
            return []
        return [json.loads(line) for line in self.path.read_text(encoding="utf-8").splitlines() if line]
    def latest(self) -> list[dict[str, Any]]:
        return self.list_all()[-100:]
 __all__ = ["TaskStore"]
@@ -1,35 +0,0 @@
 # CMD-20260709-enable-autonomous-controller
 ## Command ID
 CMD-20260709-enable-autonomous-controller
 ## Priority
 P0
 ## Mode
 autonomous
 ## Target
 Primary: OpenClaw
 Supporting: Hermes
 Review: Codex
 ## Objective
 Enable autonomous controller for AIOS GitOps:
 - Auto-detect issues
 - Auto-diagnose
 - Auto-generate fixes
 - Auto-test
 - Auto-create patches
 - Auto-request approval
 ## Preconditions
 - Agent Command Plane initialized (CMD-BOOTSTRAP-001)
 - Git Actions write permission fixed (CMD-20260708)
 - Audit plane enabled (CMD-20260708)
 - Notification service enabled (CMD-20260708)
 ## Completion Criteria
 - Autonomous controller service running
 - Can detect, diagnose, fix, and report
 - Pipeline: detect → diagnose → fix → test → patch → approve → commit → notify
@@ -1,7 +0,0 @@
 {
  "nid": "n-1783527185",
  "title": "⚠️ Issue Detected: branch_sync",
  "body": "10 branches missing from Gitea DB",
  "urgency": "warning",
  "ts": "2026-07-09T00:13:05+0800"
 }
@@ -1,7 +0,0 @@
 {
  "nid": "n-1783529900",
  "title": "⚠️ Issue Detected: branch_sync",
  "body": "10 branches missing from Gitea DB",
  "urgency": "warning",
  "ts": "2026-07-09T00:58:20+0800"
 }
@@ -1,7 +0,0 @@
 {
  "nid": "n-1783529914",
  "title": "⚠️ Issue Detected: branch_sync",
  "body": "10 branches missing from Gitea DB",
  "urgency": "warning",
  "ts": "2026-07-09T00:58:34+0800"
 }
@@ -1,7 +0,0 @@
 {
  "nid": "n-1783530514",
  "title": "⚠️ Issue Detected: branch_sync",
  "body": "10 branches missing from Gitea DB",
  "urgency": "warning",
  "ts": "2026-07-09T01:08:34+0800"
 }
@@ -1,7 +0,0 @@
 {
  "nid": "n-1783531114",
  "title": "⚠️ Issue Detected: branch_sync",
  "body": "10 branches missing from Gitea DB",
  "urgency": "warning",
  "ts": "2026-07-09T01:18:34+0800"
 }
@@ -1,7 +0,0 @@
 {
  "nid": "n-1783531714",
  "title": "⚠️ Issue Detected: branch_sync",
  "body": "10 branches missing from Gitea DB",
  "urgency": "warning",
  "ts": "2026-07-09T01:28:34+0800"
 }
@@ -1,7 +0,0 @@
 {
  "nid": "n-1783532314",
  "title": "⚠️ Issue Detected: branch_sync",
  "body": "10 branches missing from Gitea DB",
  "urgency": "warning",
  "ts": "2026-07-09T01:38:34+0800"
 }
@@ -1,7 +0,0 @@
 {
  "nid": "n-1783532914",
  "title": "⚠️ Issue Detected: branch_sync",
  "body": "10 branches missing from Gitea DB",
  "urgency": "warning",
  "ts": "2026-07-09T01:48:34+0800"
 }
@@ -1,7 +0,0 @@
 {
  "nid": "n-1783533514",
  "title": "⚠️ Issue Detected: branch_sync",
  "body": "10 branches missing from Gitea DB",
  "urgency": "warning",
  "ts": "2026-07-09T01:58:34+0800"
 }
@@ -1,7 +0,0 @@
 {
  "nid": "n-1783534114",
  "title": "⚠️ Issue Detected: branch_sync",
  "body": "10 branches missing from Gitea DB",
  "urgency": "warning",
  "ts": "2026-07-09T02:08:34+0800"
 }
@@ -1,16 +0,0 @@
 {
  "patch_id": "patch-1783527185",
  "created": "2026-07-09T00:13:05+0800",
  "check": "branch_sync",
  "severity": "high",
  "diagnosis": "10 branches missing from Gitea DB",
  "branches": [
    "patch/agent-event-bus-unified",
    "test/alias-write-verify",
    "patch/V1R2C3-tag-execute-idempotent-permission",
    "patch/gitops-write-auth-smoke-20260706T115230Z",
    "patch/gpt-actions-auth-live-20260708"
  ],
  "stuck": [],
  "fix": "sync_branches"
 }
@@ -1,16 +0,0 @@
 {
  "patch_id": "patch-1783529900",
  "created": "2026-07-09T00:58:20+0800",
  "check": "branch_sync",
  "severity": "high",
  "diagnosis": "10 branches missing from Gitea DB",
  "branches": [
    "test/alias-test-file-branch",
    "patch/codex-online-sync-acceptance-20260706T102819Z",
    "patch/V1R2C3-tag-execute-idempotent-permission",
    "patch/actions-invalid-diag-smoke-20260706T140311Z",
    "patch/V1R2C2-autonomous-recovery-loop"
  ],
  "stuck": [],
  "fix": "sync_branches"
 }
@@ -1,16 +0,0 @@
 {
  "patch_id": "patch-1783529914",
  "created": "2026-07-09T00:58:34+0800",
  "check": "branch_sync",
  "severity": "high",
  "diagnosis": "10 branches missing from Gitea DB",
  "branches": [
    "patch/codex-write-auth-smoke-20260706T115256Z",
    "patch/v1r3c0-user-portal-rbac-ops-20260708",
    "request/tag-V1R3C0-self-loop-autonomous-engine",
    "patch/fix-tag-confirmation-state-machine",
    "patch/gitops-write-auth-fix"
  ],
  "stuck": [],
  "fix": "sync_branches"
 }
@@ -1,16 +0,0 @@
 {
  "patch_id": "patch-1783530514",
  "created": "2026-07-09T01:08:34+0800",
  "check": "branch_sync",
  "severity": "high",
  "diagnosis": "10 branches missing from Gitea DB",
  "branches": [
    "patch/codex-write-auth-smoke-20260706T115256Z",
    "patch/v1r3c0-user-portal-rbac-ops-20260708",
    "request/tag-V1R3C0-self-loop-autonomous-engine",
    "patch/fix-tag-confirmation-state-machine",
    "patch/gitops-write-auth-fix"
  ],
  "stuck": [],
  "fix": "sync_branches"
 }
@@ -1,16 +0,0 @@
 {
  "patch_id": "patch-1783531114",
  "created": "2026-07-09T01:18:34+0800",
  "check": "branch_sync",
  "severity": "high",
  "diagnosis": "10 branches missing from Gitea DB",
  "branches": [
    "patch/codex-write-auth-smoke-20260706T115256Z",
    "patch/v1r3c0-user-portal-rbac-ops-20260708",
    "request/tag-V1R3C0-self-loop-autonomous-engine",
    "patch/fix-tag-confirmation-state-machine",
    "patch/gitops-write-auth-fix"
  ],
  "stuck": [],
  "fix": "sync_branches"
 }
@@ -1,16 +0,0 @@
 {
  "patch_id": "patch-1783531714",
  "created": "2026-07-09T01:28:34+0800",
  "check": "branch_sync",
  "severity": "high",
  "diagnosis": "10 branches missing from Gitea DB",
  "branches": [
    "patch/codex-write-auth-smoke-20260706T115256Z",
    "patch/v1r3c0-user-portal-rbac-ops-20260708",
    "request/tag-V1R3C0-self-loop-autonomous-engine",
    "patch/fix-tag-confirmation-state-machine",
    "patch/gitops-write-auth-fix"
  ],
  "stuck": [],
  "fix": "sync_branches"
 }
@@ -1,16 +0,0 @@
 {
  "patch_id": "patch-1783532314",
  "created": "2026-07-09T01:38:34+0800",
  "check": "branch_sync",
  "severity": "high",
  "diagnosis": "10 branches missing from Gitea DB",
  "branches": [
    "patch/codex-write-auth-smoke-20260706T115256Z",
    "patch/v1r3c0-user-portal-rbac-ops-20260708",
    "request/tag-V1R3C0-self-loop-autonomous-engine",
    "patch/fix-tag-confirmation-state-machine",
    "patch/gitops-write-auth-fix"
  ],
  "stuck": [],
  "fix": "sync_branches"
 }
--- a/Show More
+++ b/Show More
Author	SHA1	Message	Date
openclaw	9571cb66c8	Add OpenClaw deployment command for Agent GPT message plane	2026-07-08 16:33:46 +08:00
openclaw	e01ca567c3	Add Agent GPT message plane configuration	2026-07-08 16:33:39 +08:00
openclaw	e81e56bac1	Add Agent GPT message plane admin dashboard model	2026-07-08 16:33:34 +08:00
openclaw	df157d63d7	Add Agent GPT message plane Actions API	2026-07-08 16:33:30 +08:00
openclaw	a3b9a70625	Implement OpenClaw to Custom GPT message plane core	2026-07-08 16:33:25 +08:00
openclaw	fd2a7ab024	Create agent GPT message plane implementation branch	2026-07-08 16:33:16 +08:00