feat(diagnostics): bind manual reads to exact session and run evidence

docs/architecture/rfcs/harness-selection-dsh-pi-v0.md

@@ -454,6 +454,26 @@ with no diagnostic input passed into prompts or scheduler decisions. Existing
 CLI ledger reads are unbounded; do not put this command on an automatic polling
 loop before adding an owner-reviewed read budget/snapshot strategy.
 
+### Exact-binding manual readback
+
+The optional `status --expected-binding '<JSON>' --as-of <timestamp>
+--max-observation-age-ms <integer>` read binds the existing goal argument to
+one expected `provider_id`, `observer_id`, `session_id` and `run_identity`.
+The latter uses the observer stats' existing exact fields: `worker_id`,
+`model_id`, `task_id`, `environment_id`, `tools_id`, `budget_id`,
+`adapter_revision` and `observer_revision`. Tokens follow the observer identity
+schema; unknown/missing fields fail before producing a projection.
+
+This bound read requires an explicit evaluation timestamp and caller-selected
+non-negative age ceiling. Its `binding.eligible` is false for another session or
+treatment, missing/invalid integrity, future observations or an exceeded age
+ceiling; the stage is then `unknown`. It rejects ambiguity rather than hiding
+other runs by filtering the ledger. Eligibility says only that this binding's
+observation is admissible, never that its task succeeded. Unbound historical
+reads retain their prior shape. Both forms remain manual, read-only and
+unbounded ledger reads: this contract does not authorize automatic polling or
+supply the owner-reviewed read budget required for a panel.
+
 ## Qualification Plan and Stop Conditions
 
 1. **C0 adapter fidelity:** compare native execution with the managed adapter,

docs/architecture/rfcs/harness-selection-dsh-pi-v0.zh-CN.md

@@ -595,3 +595,19 @@ lane 的接收者解析、报告可先用的目标级里程碑——仍然开放
 profile、权限状态、第二执行器或工作权威，因此更丰富的回答契约不得扩大通道可读或
 可改的范围；本文也不提升任何一行的状态——M1-M4 接入里程碑与跨前端投影行仍归
 [Agent 会话执行模式](./agent-session-execution-modes-v0.zh-CN.md)。
+
+
+### 精确绑定的手动诊断读取
+
+`status --expected-binding '<JSON>' --as-of <timestamp> --max-observation-age-ms <integer>`
+在现有 goal 参数上固定 `provider_id`、`observer_id`、`session_id` 和 `run_identity`。
+后者使用 observer stats 的现有完整字段：`worker_id`、`model_id`、`task_id`、
+`environment_id`、`tools_id`、`budget_id`、`adapter_revision`、`observer_revision`。
+缺少或未知字段、非法 identity token 会拒绝输出。
+
+调用方必须给出明确评价时间及非负观测年龄上限。其他 session/treatment、
+完整性不合法、缺少观测、未来观测或超过年龄上限都会令 `binding.eligible=false`，
+并令 stage 为 `unknown`，而不是过滤掉其他运行后误称健康。eligible 只表示该绑定的
+观测可采用，不表示任务成功。不指定绑定时保留原有历史读取格式。读取仍然手动、
+只读且不限制 ledger 大小；不授权自动轮询，不替代面板必需的 owner-reviewed 读取预算，
+也不证明 C0/C1 live qualification。

loopx/capabilities/reliability_diagnostics/README.md

@@ -153,6 +153,25 @@ text; readback does not rewrite the ledger.
 | `signals` | `stall_suspected`, `repetition_suspected`, `unrecovered_error`, `event_loss`, `integrity_not_valid` |
 | `integrity` | the receipt status and reason codes |
 
+### Exact binding and freshness
+
+For a manual current-session read, pass `status --expected-binding '<JSON>'`
+with `--as-of <current timestamp>` and `--max-observation-age-ms <nonnegative integer>`.
+The JSON object has exactly `provider_id`, `observer_id`, `session_id` and
+`run_identity` (the eight existing observer-stats treatment fields above).
+Invalid input exits with code 2. Omission preserves historical projection output.
+
+The optional `binding` object contains `expected`, `eligible`, `reason_codes`
+and `max_observation_age_ms`. It refuses an empty ledger, any different
+session/provider/observer or run treatment, non-valid integrity, any future
+observation, or a latest observation older than the caller's ceiling. Refusal
+sets `stage` to `unknown`; counts and signals remain historical ledger evidence.
+Reason codes are local to this read contract: `no_observations`,
+`session_identity_mismatch`, `run_identity_mismatch`, `integrity_not_valid`,
+`observation_in_future`, and `observation_stale`. Matching evidence proves
+admissibility only, never task success. The read does not filter out conflicting
+runs, write state, call a provider, or enable polling.
+
 ## Use It
 
 ```bash

loopx/capabilities/reliability_diagnostics/README.zh-CN.md

@@ -219,3 +219,19 @@ projection 报告 `read` 上的重复、一次已恢复的错误、无 stall。
 不改 `loopx status` 首屏。适配器要求外部固定 goal/session/run identity；不会通过 Driver
 或 LoopX CLI 发现绑定。本原型也不提供 matched native/L1 执行、真实 DSH host 测量，或一次合格 C1 实跑。
 bench 工具只在 synthetic 事件流上测量 observer 组件开销，不构成 live 资格。
+
+
+### 精确绑定与新鲜度
+
+手动读取当前会话时，向 `status` 传入 `--expected-binding '<JSON>'`、显式
+`--as-of <当前时间>` 和 `--max-observation-age-ms <非负整数>`。JSON 恰好包含
+`provider_id`、`observer_id`、`session_id` 与 `run_identity`；后者复用 observer
+stats 已有的八个 treatment 字段。非法参数以退出码 2 拒绝；不传绑定时保留历史输出。
+
+新增的 `binding` 包含 `expected`、`eligible`、`reason_codes` 和年龄上限。
+空账本、任一会话／provider／observer 或运行 treatment 不匹配、非 valid 完整性、
+未来观测或最新观测超龄都会拒绝，并把 `stage` 设为 `unknown`；计数与信号仍是历史
+账本证据。原因码为本地读取合同的 `no_observations`、`session_identity_mismatch`、
+`run_identity_mismatch`、`integrity_not_valid`、`observation_in_future`、
+`observation_stale`。匹配只证明观测可采纳，不证明任务成功。读取不会过滤冲突运行、
+写状态、调用 provider 或启用轮询。

loopx/capabilities/reliability_diagnostics/projection.py

@@ -8,18 +8,21 @@
 
 from __future__ import annotations
 
+from collections.abc import Mapping
 from dataclasses import dataclass, field
 from enum import StrEnum
 from fractions import Fraction
 from typing import Any
 
 from .envelope import (
     CAPABILITY_ID,
+    IDENTITY_TOKEN_PATTERN,
     ObserverEnvelope,
     ObserverEventKind,
     observed_at_microseconds,
     parse_observed_at,
 )
+from .intake import normalize_observer_run_identity
 from .receipt import LedgerReading, build_integrity_receipt
 
 DIAGNOSTIC_PROJECTION_SCHEMA_VERSION = "reliability_diagnostic_projection_v0"
@@ -47,6 +50,16 @@ class DiagnosticSignal(StrEnum):
     INTEGRITY_NOT_VALID = "integrity_not_valid"
 
 
+class BindingIneligibility(StrEnum):
+    """Owner-local reasons for refusing a bound diagnostic read, not runtime state."""
+    NO_OBSERVATIONS = "no_observations"
+    SESSION_IDENTITY_MISMATCH = "session_identity_mismatch"
+    RUN_IDENTITY_MISMATCH = "run_identity_mismatch"
+    INTEGRITY_NOT_VALID = "integrity_not_valid"
+    OBSERVATION_IN_FUTURE = "observation_in_future"
+    OBSERVATION_STALE = "observation_stale"
+
+
 _ACTIVE_STAGES = frozenset({DiagnosticStage.RUNNING, DiagnosticStage.TOOL_RUNNING})
 
 
@@ -192,7 +205,38 @@ def build_diagnostic_projection(
     as_of: str | None = None,
     stall_threshold_ms: int = DEFAULT_STALL_THRESHOLD_MS,
     repetition_threshold: int = DEFAULT_REPETITION_THRESHOLD,
+    expected_binding: Mapping[str, Any] | None = None,
+    max_observation_age_ms: int | None = None,
 ) -> dict[str, Any]:
+    binding = None
+    if expected_binding is not None:
+        fields = {"provider_id", "observer_id", "session_id", "run_identity"}
+        if not isinstance(expected_binding, Mapping) or set(expected_binding) != fields:
+            raise ValueError(
+                "expected_binding requires provider_id, observer_id, session_id and run_identity"
+            )
+        for key in fields - {"run_identity"}:
+            value = expected_binding[key]
+            if not isinstance(value, str) or not IDENTITY_TOKEN_PATTERN.fullmatch(
+                value
+            ):
+                raise ValueError(f"expected_binding.{key} must be an identity token")
+        identity = normalize_observer_run_identity(expected_binding["run_identity"])
+        binding = {**expected_binding, "run_identity": identity.as_dict()}
+        if as_of is None:
+            raise ValueError(
+                "expected_binding requires an explicit as_of evaluation time"
+            )
+        if (
+            isinstance(max_observation_age_ms, bool)
+            or not isinstance(max_observation_age_ms, int)
+            or max_observation_age_ms < 0
+        ):
+            raise ValueError(
+                "expected_binding requires a non-negative max_observation_age_ms"
+            )
+    elif max_observation_age_ms is not None:
+        raise ValueError("max_observation_age_ms requires expected_binding")
     if as_of is not None:
         message = "as_of must be a timezone-aware ISO-8601 timestamp"
         try:
@@ -227,7 +271,7 @@ def build_diagnostic_projection(
         receipt=receipt,
     )
 
-    return {
+    result = {
         "schema_version": DIAGNOSTIC_PROJECTION_SCHEMA_VERSION,
         "capability_id": CAPABILITY_ID,
         "goal_id": reading.goal_id,
@@ -269,3 +313,46 @@ def build_diagnostic_projection(
             "as_of": effective_as_of,
         },
     }
+    if binding is not None:
+        reasons: list[str] = []
+        if not envelopes:
+            reasons.append(BindingIneligibility.NO_OBSERVATIONS.value)
+        if any(
+            item.provider_id != binding["provider_id"]
+            or item.observer_id != binding["observer_id"]
+            or item.session_id != binding["session_id"]
+            for item in envelopes
+        ):
+            reasons.append(BindingIneligibility.SESSION_IDENTITY_MISMATCH.value)
+        stats = list(reading.stats.values())
+        if len(stats) != 1 or any(
+            item.provider_id != binding["provider_id"]
+            or item.observer_id != binding["observer_id"]
+            or item.run_identity.as_dict() != binding["run_identity"]
+            for item in stats
+        ):
+            reasons.append(BindingIneligibility.RUN_IDENTITY_MISMATCH.value)
+        if receipt["status"] != "valid":
+            reasons.append(BindingIneligibility.INTEGRITY_NOT_VALID.value)
+        # Compare instants; textual timestamp ordering can mix UTC offsets.
+        observation_ages_us = [
+            observed_at_microseconds(effective_as_of)
+            - observed_at_microseconds(item.observed_at)
+            for item in envelopes
+        ]
+        if any(age < 0 for age in observation_ages_us):
+            reasons.append(BindingIneligibility.OBSERVATION_IN_FUTURE.value)
+        elif (
+            observation_ages_us
+            and min(observation_ages_us) > max_observation_age_ms * 1000
+        ):
+            reasons.append(BindingIneligibility.OBSERVATION_STALE.value)
+        result["binding"] = {
+            "expected": binding,
+            "eligible": not reasons,
+            "reason_codes": reasons,
+            "max_observation_age_ms": max_observation_age_ms,
+        }
+        if reasons:
+            result["stage"] = DiagnosticStage.UNKNOWN.value
+    return result

loopx/cli_commands/reliability_diagnostics.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import argparse
+import json
 import sys
 from collections.abc import Callable
 from dataclasses import dataclass
@@ -89,6 +90,11 @@ def register_reliability_diagnostics_commands(
         action="store_true",
         help="Include integrity evidence from the same ledger read; grants no control authority.",
     )
+    status.add_argument(
+        "--expected-binding",
+        help="JSON object pinning provider_id, observer_id, session_id and typed run_identity; requires --as-of and --max-observation-age-ms.",
+    )
+    status.add_argument("--max-observation-age-ms", type=int)
 
 
 def _render(payload: dict[str, Any]) -> str:
@@ -104,6 +110,7 @@ def _render(payload: dict[str, Any]) -> str:
             )
             for key in (
                 "stage",
+                "binding",
                 "signals",
                 "reason_codes",
                 "lost_event_count",
@@ -254,8 +261,16 @@ def handle_reliability_diagnostics_command(
             if command == "receipt":
                 payload["receipt"] = build_integrity_receipt(reading)
             else:
+                expected_binding = None
+                if args.expected_binding is not None:
+                    expected_binding = json.loads(args.expected_binding)
+                    if not isinstance(expected_binding, dict):
+                        raise ValueError("expected_binding must be a JSON object")
                 payload["projection"] = build_diagnostic_projection(
-                    reading, as_of=args.as_of
+                    reading,
+                    as_of=args.as_of,
+                    expected_binding=expected_binding,
+                    max_observation_age_ms=args.max_observation_age_ms,
                 )
                 if args.with_receipt:
                     payload["receipt"] = build_integrity_receipt(reading)