feat(reliability-diagnostics): harden the default-off DSH shadow observer

docs/architecture/rfcs/long-running-agent-reliability-diagnostics-governed-delivery-v0.md

@@ -524,6 +524,16 @@ source. Prove the no-outbound-control invariant and bounded failure behavior.
 **Exit:** C0 adapter fidelity plus an eligible C1 observer run; public/private
 boundary and overhead are reported; no production authority exists.
 
+**Checkpoint (2026-09):** the contract half of P0 exists as the default-off
+built-in capability `reliability-diagnostics` with the extension provider
+`dsh-session-events` in `packages/dsh-loopx-plugin`: provider-neutral
+envelope and stats records, integrity receipt, read-only diagnostic
+projection, a deterministic DSH-shaped fixture, and producer-side
+public-safety rejection before the first ledger append. Still open before P0
+exit: an eligible C1 observer run on a real `dsh` session, the reported
+overhead measurement, and the ledger retention and deletion profile from
+decision 4 below.
+
 ### P1 — Benchmark-qualified diagnostic pilot
 
 Run matched native and L1 arms on at least one suitable benchmark family and
@@ -594,6 +604,13 @@ required before making a stronger commercial claim.
    pilot: software delivery, security/SRE, or research/AI4S?
 2. Which event source and harness should define the P0 shadow-observer
    conformance fixture?
+   **Decided (2026-09): DeepSeek Harness (`dsh`) session events.** LoopX
+   already ships a typed `dsh` Turn host and a same-session plugin whose
+   read-only `session/event`, `agent/status`, `agent/error`, and
+   `session/disposed` hooks let the observer be proven non-interfering inside
+   an existing packaged boundary. Pi remains the comparison candidate; the
+   harness-selection evaluation shared with the Desktop Execution Frontends
+   RFC is a follow-up deliverable and will be recorded here.
 3. Should the first two-to-four-week offer stop at L1 diagnostics by default,
    or include an optional L2 advisory week before any L3 seam?
 4. Which data-retention, deletion, and support profiles belong in the first
@@ -613,6 +630,12 @@ required before making a stronger commercial claim.
   owns benchmark truth, matched arms, C0–C4 evidence, and research integrity.
 - [Agent Management Observability MVP](../../product/surfaces/agent-management-observability-mvp.md)
   defines the read-only projection posture reused by L1/L2 operator surfaces.
+- [Desktop Execution Frontends](./desktop-execution-frontends-v0.md) defines
+  Mode B, the Managed Agent Runtime in which LoopX Desktop launches and
+  supervises Pi or `dsh`. The L1 shadow observer is the passive diagnostic
+  layer under that mode's Desktop-owned runtime supervisor: its integrity
+  receipt and read-only projection are inputs the supervisor may project, and
+  the observer acquires none of the supervisor's authority.
 - [Shared Goal Authority and State Provider](./shared-goal-authority-state-provider-v0.md)
   defines the authority/provider boundary required only when L3/L4 uses shared
   coordination.

docs/architecture/rfcs/long-running-agent-reliability-diagnostics-governed-delivery-v0.zh-CN.md

@@ -447,6 +447,13 @@ to pay 的证明。
 **Exit：** C0 adapter fidelity 加一条 eligible C1 observer run；报告 public/private boundary 与
 overhead；不存在 production authority。
 
+**Checkpoint（2026-09）：** P0 的 contract 部分已以默认关闭的 built-in capability
+`reliability-diagnostics` 与 extension provider `dsh-session-events`（位于
+`packages/dsh-loopx-plugin`）落地：provider-neutral envelope 与 stats record、integrity receipt、
+read-only diagnostic projection、deterministic DSH-shaped fixture，以及首次写入 ledger 之前的
+producer 侧 public-safety 拒绝。P0 exit 之前仍未完成：在真实 `dsh` session 上的 eligible C1
+observer run、overhead 测量报告，以及下文 decision 4 的 ledger retention 与 deletion profile。
+
 ### P1 — Benchmark-qualified diagnostic pilot
 
 在至少一个合适 benchmark family 和一个 non-benchmark rehearsal 上运行 matched native/L1 arm。
@@ -506,6 +513,10 @@ advantage 与 sustainable delivery evidence。
 1. 首个产品 pilot 应选择哪个 initial ICP 与 reference workflow：software delivery、security/SRE，
    还是 research/AI4S？
 2. 哪个 event source 与 harness 应定义 P0 shadow-observer conformance fixture？
+   **已决定（2026-09）：DeepSeek Harness（`dsh`）session events。** LoopX 已有 typed `dsh`
+   Turn host 与 same-session plugin，其只读 `session/event`、`agent/status`、`agent/error`、
+   `session/disposed` hook 让 observer 能在既有打包边界内被证明 non-interfering。Pi 仍是
+   对比候选；与 Desktop Execution Frontends RFC 共享的 harness 选型评估是后续交付物，结论将记录在此。
 3. 第一份两到四周 offer 默认应停在 L1 diagnostic，还是在进入任何 L3 seam 前增加可选 L2 advisory week？
 4. 第一份 local/private/BYOC deployment pack 应包含哪些 data-retention、deletion 与 support profile？
 5. 第一份 promotion packet 必须使用哪个 benchmark family 与 non-benchmark canary？
@@ -520,6 +531,10 @@ advantage 与 sustainable delivery evidence。
   拥有 benchmark truth、matched arm、C0–C4 evidence 与 research integrity。
 - [Agent Management Observability MVP](../../product/surfaces/agent-management-observability-mvp.md)
   定义 L1/L2 operator surface 复用的 read-only projection posture。
+- [Desktop Execution Frontends](./desktop-execution-frontends-v0.zh-CN.md) 定义 Mode B，即由 LoopX
+  Desktop 启动并监督 Pi 或 `dsh` 的 Managed Agent Runtime。L1 shadow observer 是该模式下
+  Desktop-owned runtime supervisor 之下的被动诊断层：其 integrity receipt 与 read-only projection
+  是 supervisor 可以投影的输入，observer 本身不获得 supervisor 的任何 authority。
 - [Shared Goal Authority 与 State Provider](./shared-goal-authority-state-provider-v0.zh-CN.md)
   定义只有在 L3/L4 使用 shared coordination 时才需要的 authority/provider boundary。
 - [TypeScript Control-Plane Migration](./typescript-control-plane-migration-v0.zh-CN.md)

examples/reliability_diagnostics/dsh-shadow-observer-fixture-smoke.py

@@ -0,0 +1,167 @@
+#!/usr/bin/env python3
+"""Prove the L1 shadow-observer contract on the deterministic DSH-shaped fixture.
+
+Assertions come from the reliability-diagnostics design contract (RFC §3.1
+non-interference, §7.4 integrity receipt): no outbound control path, bounded
+and counted loss, visible clock uncertainty, raw material never persisted, and
+a read-only projection with no authority. The smoke also proves the CLI
+readback path round-trips the same ledger.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+import tempfile
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(REPO_ROOT))
+
+from loopx.capabilities.reliability_diagnostics import (  # noqa: E402
+    CONTROL_FIELD_FAMILIES,
+    ENVELOPE_FIELDS,
+    FIXTURE_GOAL_ID,
+    RAW_MATERIAL_FIELD_FAMILIES,
+    dsh_fixture_records,
+    run_dsh_fixture,
+)
+from loopx.capabilities.reliability_diagnostics.fixture import (  # noqa: E402
+    FIXTURE_BUFFER_BOUND,
+    FIXTURE_UNCERTAIN_CLOCK_MS,
+)
+
+
+def run_cli(*args: str, runtime_root: Path, stdin: str | None = None) -> dict[str, object]:
+    completed = subprocess.run(
+        [
+            sys.executable,
+            "-m",
+            "loopx.cli",
+            "--runtime-root",
+            str(runtime_root),
+            "--format",
+            "json",
+            "reliability-diagnostics",
+            *args,
+        ],
+        cwd=REPO_ROOT,
+        env={**os.environ, "PYTHONPATH": str(REPO_ROOT)},
+        input=stdin,
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    assert completed.returncode == 0, completed.stderr or completed.stdout
+    return json.loads(completed.stdout)
+
+
+def assert_no_outbound_control(receipt: dict, projection: dict) -> None:
+    assert receipt["outbound_endpoints"] == [], receipt
+    assert receipt["observation_entered_worker_context"] is False, receipt
+    assert receipt["observation_entered_scheduler_inputs"] is False, receipt
+    assert projection["mode"] == "read_only", projection
+    assert projection["authority"] == "none", projection
+    assert projection["worker_influence"] == "none", projection
+    flattened = {name.replace("_", "") for name in ENVELOPE_FIELDS}
+    assert not flattened & CONTROL_FIELD_FAMILIES
+    assert not flattened & RAW_MATERIAL_FIELD_FAMILIES
+
+
+def assert_bounded_failure(result: dict) -> None:
+    receipt = result["receipt"]
+    stats = result["stats"]
+    fixture_records = dsh_fixture_records()
+    # Every fixture record is observed once, then accepted, rejected, or dropped.
+    assert len(fixture_records) == receipt["observed_event_count"], (
+        len(fixture_records),
+        receipt,
+    )
+    assert receipt["observed_event_count"] == (
+        receipt["accepted_event_count"]
+        + receipt["rejected_event_count"]
+        + receipt["backpressure_drop_count"]
+    ), receipt
+    assert stats["buffer_bound"] == FIXTURE_BUFFER_BOUND
+    assert receipt["backpressure_drop_count"] == 3, receipt
+    # Sequence 10 and the rejected sequence 19 are visible as gaps; trailing
+    # drops are visible only through the stats record.
+    assert receipt["lost_event_count"] == 2, receipt
+    assert receipt["clock"]["max_uncertainty_ms"] == FIXTURE_UNCERTAIN_CLOCK_MS
+    assert receipt["rejected_by_reason"] == {"raw_material_field_rejected": 1}, receipt
+    assert receipt["observer_failure_count"] == 0
+    assert receipt["persisted_event_count"] == receipt["accepted_event_count"]
+    assert receipt["event_sources"] == [
+        "session/created",
+        "session/disposed",
+        "session/event",
+    ]
+    assert receipt["status"] == "degraded", receipt
+    assert set(receipt["reason_codes"]) == {
+        "sequence_gap",
+        "backpressure_drop",
+        "raw_material_rejected",
+        "clock_uncertainty_exceeded",
+    }, receipt
+    assert "transcript" not in json.dumps(result["ledger_records"])
+    assert "protected task content" not in json.dumps(result)
+
+
+def assert_projection_signals(projection: dict) -> None:
+    assert projection["repetition"] == {
+        "detected": True,
+        "threshold": 3,
+        "longest_tool_run": 3,
+        "tool_name": "read",
+    }, projection
+    assert projection["recovery"] == {
+        "error_count": 1,
+        "recovered_error_count": 1,
+        "unrecovered_error_count": 0,
+    }, projection
+    assert projection["stall"]["detected"] is False, projection
+    assert set(projection["signals"]) == {
+        "repetition_suspected",
+        "event_loss",
+        "integrity_not_valid",
+    }, projection
+    assert projection["integrity"]["status"] == "degraded"
+
+
+def assert_cli_readback(result: dict) -> None:
+    with tempfile.TemporaryDirectory() as tmp:
+        runtime_root = Path(tmp)
+        ndjson = "\n".join(json.dumps(record) for record in result["ledger_records"]) + "\n"
+        ingest = run_cli("ingest", "--goal-id", FIXTURE_GOAL_ID, "--input", "-", runtime_root=runtime_root, stdin=ndjson)
+        assert ingest["ok"] is True, ingest
+        assert ingest["appended_record_count"] == len(result["ledger_records"]), ingest
+        assert ingest["ledger_ref"] == f"reliability_diagnostics/{FIXTURE_GOAL_ID}.ndjson"
+        assert str(runtime_root) not in json.dumps(ingest)
+
+        receipt = run_cli("receipt", "--goal-id", FIXTURE_GOAL_ID, runtime_root=runtime_root)
+        assert receipt["receipt"] == result["receipt"], receipt
+        status = run_cli("status", "--goal-id", FIXTURE_GOAL_ID, runtime_root=runtime_root)
+        assert status["projection"] == result["projection"], status
+
+        # A refused input leaves a durable invalid gate record in the ledger.
+        poisoned = dict(result["ledger_records"][0])
+        poisoned["command"] = {"kind": "stop"}
+        rejected = run_cli("ingest", "--goal-id", FIXTURE_GOAL_ID, "--input", "-", runtime_root=runtime_root, stdin=json.dumps(poisoned) + "\n")
+        assert rejected["rejected_by_reason"] == {"control_field_rejected": 1}, rejected
+        assert run_cli("receipt", "--goal-id", FIXTURE_GOAL_ID, runtime_root=runtime_root)["receipt"]["status"] == "invalid"
+
+
+def main() -> int:
+    result = run_dsh_fixture()
+    assert_no_outbound_control(result["receipt"], result["projection"])
+    assert_bounded_failure(result)
+    assert_projection_signals(result["projection"])
+    assert_cli_readback(result)
+    print("reliability-diagnostics dsh-shadow-observer-fixture-smoke: ok")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

loopx/capabilities/README.md

@@ -74,6 +74,7 @@ availability and maturity in the installed release.
 | Turn public/private content signals into reviewable source, angle, draft, feedback, and publish-gate packets | [Content Operations](content_ops/README.md) |
 | Inventory, archive, migrate, and rerank a material store without losing raw source authority | [Material Lifecycle](material_lifecycle/README.md) ([中文](material_lifecycle/README.zh-CN.md)) |
 | Inspect compatibility routes for public-safe external-value intake while callers migrate to outcome-owned capabilities | [Value Connectors](value_connectors/README.md) |
+| Observe a long-running harness session one-way and read back an integrity receipt and stall/repetition/recovery projection with no runtime authority | [Reliability Diagnostics](reliability_diagnostics/README.md) ([中文](reliability_diagnostics/README.zh-CN.md)) |
 
 ## Contributor Navigation And Ownership
 

loopx/capabilities/catalog.py

@@ -26,6 +26,7 @@
 from .deep_research.catalog_entry import DEEP_RESEARCH_CATALOG_ENTRY
 from .public_safe_outbound.catalog_entry import PUBLIC_SAFE_OUTBOUND_CATALOG_ENTRY
 from .connector_registry.catalog_entry import CONNECTOR_REGISTRY_CATALOG_ENTRY
+from .reliability_diagnostics.catalog_entry import RELIABILITY_DIAGNOSTICS_CATALOG_ENTRY
 from .registry import CapabilityRegistry
 
 CAPABILITY_CATALOG_SCHEMA_VERSION = "loopx_capability_catalog_v0"
@@ -52,6 +53,7 @@
     DEEP_RESEARCH_CATALOG_ENTRY,
     PUBLIC_SAFE_OUTBOUND_CATALOG_ENTRY,
     CONNECTOR_REGISTRY_CATALOG_ENTRY,
+    RELIABILITY_DIAGNOSTICS_CATALOG_ENTRY,
 )
 # Preserve the original import surface while routing all reads through the registry.
 CAPABILITIES = BUILTIN_CAPABILITIES
@@ -78,6 +80,44 @@ def _summary(record: Mapping[str, Any]) -> dict[str, Any]:
     }
 
 
+def _register_declared_extension_providers(
+    registry: CapabilityRegistry,
+    record: Mapping[str, object],
+) -> None:
+    """Declare extension-delivered implementations named by a builtin entry.
+
+    A provider distributed outside the Python extension lifecycle (for example
+    an npm plugin) has no manifest or state file. The owning capability
+    declares it so the catalog reports `declared=true` and
+    `installed=enabled=ready=false` until a real lifecycle proves otherwise.
+    """
+
+    for implementation in record.get("implementation_providers") or []:
+        if not isinstance(implementation, Mapping) or implementation.get("origin") != "extension":
+            continue
+        provider_id = str(implementation["provider_id"])
+        if not any(provider["id"] == provider_id for provider in registry.providers()):
+            registry.register_provider(
+                {
+                    "id": provider_id,
+                    "origin": "extension",
+                    "declared": True,
+                    "installed": False,
+                    "enabled": False,
+                    "ready": False,
+                }
+            )
+        registry.register_implementation(
+            {
+                "capability_id": record["id"],
+                "provider_id": provider_id,
+                "protocol": implementation["protocol"],
+                "package": implementation.get("package"),
+                "status": implementation.get("status"),
+            }
+        )
+
+
 def build_capability_registry(
     extension_manifest_paths: Iterable[str | Path] = (),
     *,
@@ -96,6 +136,8 @@ def build_capability_registry(
     )
     for record in BUILTIN_CAPABILITIES:
         registry.register_capability(record)
+    for record in BUILTIN_CAPABILITIES:
+        _register_declared_extension_providers(registry, record)
     for manifest in extension_catalog_entries(
         extension_manifest_paths,
         state_file=extension_state_file,