From 5a1c3bac5f4220d36fd4b7d7f0274f803eeafb0c Mon Sep 17 00:00:00 2001 From: Dante Date: Mon, 26 Oct 2026 09:45:03 +0800 Subject: [PATCH 1/2] feat: bind manual diagnostics to exact session and run evidence --- .../rfcs/harness-selection-dsh-pi-v0.md | 20 +++ .../rfcs/harness-selection-dsh-pi-v0.zh-CN.md | 16 +++ .../reliability_diagnostics/projection.py | 89 +++++++++++- loopx/cli_commands/reliability_diagnostics.py | 17 ++- .../test_reliability_diagnostics.py | 132 ++++++++++++++++++ .../test_reliability_diagnostics_readback.py | 88 ++++++++++++ 6 files changed, 360 insertions(+), 2 deletions(-) diff --git a/docs/architecture/rfcs/harness-selection-dsh-pi-v0.md b/docs/architecture/rfcs/harness-selection-dsh-pi-v0.md index 20a7809a75..a8184bb7a8 100644 --- a/docs/architecture/rfcs/harness-selection-dsh-pi-v0.md +++ b/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 '' --as-of +--max-observation-age-ms ` 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, diff --git a/docs/architecture/rfcs/harness-selection-dsh-pi-v0.zh-CN.md b/docs/architecture/rfcs/harness-selection-dsh-pi-v0.zh-CN.md index cc799851d2..4517444e7d 100644 --- a/docs/architecture/rfcs/harness-selection-dsh-pi-v0.zh-CN.md +++ b/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 '' --as-of --max-observation-age-ms ` +在现有 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。 diff --git a/loopx/capabilities/reliability_diagnostics/projection.py b/loopx/capabilities/reliability_diagnostics/projection.py index 96921a082e..761e099da9 100644 --- a/loopx/capabilities/reliability_diagnostics/projection.py +++ b/loopx/capabilities/reliability_diagnostics/projection.py @@ -8,6 +8,7 @@ from __future__ import annotations +from collections.abc import Mapping from dataclasses import dataclass, field from enum import StrEnum from fractions import Fraction @@ -15,11 +16,13 @@ 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 diff --git a/loopx/cli_commands/reliability_diagnostics.py b/loopx/cli_commands/reliability_diagnostics.py index b5eb11610f..25048223f8 100644 --- a/loopx/cli_commands/reliability_diagnostics.py +++ b/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) diff --git a/tests/capabilities/test_reliability_diagnostics.py b/tests/capabilities/test_reliability_diagnostics.py index f1c0bcb417..50d2b9704a 100644 --- a/tests/capabilities/test_reliability_diagnostics.py +++ b/tests/capabilities/test_reliability_diagnostics.py @@ -913,3 +913,135 @@ def test_dsh_fixture_exercises_every_contract_hazard_and_stays_degraded() -> Non + receipt["rejected_event_count"] + receipt["backpressure_drop_count"] ) + +def expected_binding(): + return { + "provider_id": DSH_PROVIDER_ID, + "observer_id": OBSERVER, + "session_id": SESSION, + "run_identity": RUN_IDENTITY.as_dict(), + } + + +def test_exact_binding_admits_only_fresh_matching_passive_evidence(): + result = projection_for( + envelope(0), + stats(), + expected_binding=expected_binding(), + as_of=at(1), + max_observation_age_ms=1000, + ) + assert result["binding"]["eligible"] is True + assert result["binding"]["reason_codes"] == [] + assert result["authority"] == "none" + + +@pytest.mark.parametrize( + "change,reason", + [ + ({"session_id": "another-session"}, "session_identity_mismatch"), + ({"observer_id": "another-observer"}, "run_identity_mismatch"), + ( + {"run_identity": {**RUN_IDENTITY.as_dict(), "task_id": "another-task"}}, + "run_identity_mismatch", + ), + ], +) +def test_exact_binding_refuses_identity_mismatch(change, reason): + result = projection_for( + envelope(0), + stats(), + expected_binding={**expected_binding(), **change}, + as_of=at(1), + max_observation_age_ms=1000, + ) + assert result["binding"]["eligible"] is False + assert reason in result["binding"]["reason_codes"] + assert result["stage"] == "unknown" + + +@pytest.mark.parametrize( + "now,reason", + [ + (at(2), "observation_stale"), + ("2026-09-01T09:59:59+00:00", "observation_in_future"), + ], +) +def test_exact_binding_refuses_old_or_future_observations(now, reason): + result = projection_for( + envelope(0), + stats(), + expected_binding=expected_binding(), + as_of=now, + max_observation_age_ms=1000, + ) + assert result["binding"]["eligible"] is False + assert reason in result["binding"]["reason_codes"] + + +def test_exact_binding_checks_future_instants_across_clock_offsets(): + result = projection_for( + envelope(0, observed_at="2026-09-01T09:30:00-01:00"), + envelope(1, observed_at="2026-09-01T10:00:00+00:00"), + stats(accepted_event_count=2), + expected_binding=expected_binding(), + as_of="2026-09-01T10:00:01+00:00", + max_observation_age_ms=1000, + ) + assert "observation_in_future" in result["binding"]["reason_codes"] + assert result["stage"] == "unknown" + + +def test_exact_binding_cannot_hide_another_session_or_invalid_integrity(): + result = projection_for( + envelope(0), + envelope(0, session_id="another-session"), + stats(accepted_event_count=2), + expected_binding=expected_binding(), + as_of=at(1), + max_observation_age_ms=1000, + ) + assert "session_identity_mismatch" in result["binding"]["reason_codes"] + invalid = projection_for( + envelope(0), + expected_binding=expected_binding(), + as_of=at(1), + max_observation_age_ms=1000, + ) + assert "integrity_not_valid" in invalid["binding"]["reason_codes"] + assert invalid["stage"] == "unknown" + + +@pytest.mark.parametrize( + "kwargs", + [ + {"max_observation_age_ms": 1000}, + {"as_of": at(1)}, + {"as_of": at(1), "max_observation_age_ms": -1}, + {"as_of": at(1), "max_observation_age_ms": True}, + ], +) +def test_exact_binding_requires_explicit_clock_and_age_bound(kwargs): + with pytest.raises(ValueError): + projection_for( + envelope(0), stats(), expected_binding=expected_binding(), **kwargs + ) + + +@pytest.mark.parametrize( + "observed,now,limit,reason", + [ + ("2026-09-01T10:00:00.0000001Z", "2026-09-01T10:00:00Z", 0, + "observation_in_future"), + ("2026-09-01T10:00:00Z", "2026-09-01T10:00:01.0000001Z", 1000, + "observation_stale"), + ], +) +def test_exact_binding_preserves_fractional_timestamp_boundaries(observed, now, limit, reason): + result = projection_for( + envelope(0, observed_at=observed), stats(), expected_binding=expected_binding(), + as_of=now, max_observation_age_ms=limit, + ) + assert result["binding"]["eligible"] is False + assert reason in result["binding"]["reason_codes"] + assert result["stage"] == "unknown" diff --git a/tests/capabilities/test_reliability_diagnostics_readback.py b/tests/capabilities/test_reliability_diagnostics_readback.py index 05eba0aec4..b120fbcf2a 100644 --- a/tests/capabilities/test_reliability_diagnostics_readback.py +++ b/tests/capabilities/test_reliability_diagnostics_readback.py @@ -114,3 +114,91 @@ def test_explicit_as_of_validation_is_independent_of_ledger( assert "as_of must be a timezone-aware ISO-8601 timestamp" in result.stderr assert not result.stdout assert (path.read_bytes() if path.exists() else None) == before + +def test_cli_exact_binding_is_fail_closed_and_read_only(tmp_path): + records = run_dsh_fixture()["ledger_records"] + stats = next(row for row in records if "run_identity" in row) + envelope = next(row for row in records if "session_id" in row) + binding = { + "provider_id": stats["provider_id"], + "observer_id": stats["observer_id"], + "session_id": envelope["session_id"], + "run_identity": stats["run_identity"], + } + path = ledger_path(tmp_path, FIXTURE_GOAL_ID) + append_ledger_records(path, records) + before = path.read_bytes() + flags = [ + "--expected-binding", + json.dumps(binding), + "--max-observation-age-ms", + "1000", + ] + result = run_status(tmp_path, *flags, "--with-receipt") + assert result["projection"]["binding"]["eligible"] is False + assert "integrity_not_valid" in result["projection"]["binding"]["reason_codes"] + assert result["projection"]["stage"] == "unknown" + assert result["receipt"]["observation_entered_scheduler_inputs"] is False + assert path.read_bytes() == before + historical = run_status(tmp_path, *flags, as_of=None, raw=True) + assert historical.returncode == 2 + assert "explicit as_of" in historical.stderr + assert not historical.stdout + assert path.read_bytes() == before + + +def test_cli_exact_binding_matching_single_run_and_invalid_input(tmp_path): + records = run_dsh_fixture()["ledger_records"] + event = next(row for row in records if "session_id" in row) + event = {**event, "sequence": 0} + stats = next(row for row in records if "run_identity" in row) + stats = { + **stats, + "observed_event_count": 1, + "accepted_event_count": 1, + "rejected_event_count": 0, + "rejected_by_reason": {}, + "backpressure_drop_count": 0, + "observer_failure_count": 0, + } + binding = { + "provider_id": stats["provider_id"], + "observer_id": stats["observer_id"], + "session_id": event["session_id"], + "run_identity": stats["run_identity"], + } + path = ledger_path(tmp_path, FIXTURE_GOAL_ID) + append_ledger_records(path, [event, stats]) + before = { + str(p.relative_to(tmp_path)): p.read_bytes() + for p in tmp_path.rglob("*") + if p.is_file() + } + result = run_status( + tmp_path, + "--expected-binding", + json.dumps(binding), + "--max-observation-age-ms", + "0", + as_of=event["observed_at"], + ) + assert result["projection"]["binding"]["eligible"] is True + assert result["projection"]["binding"]["reason_codes"] == [] + for invalid in ("null", "[]", "{}", "not-json"): + failure = run_status( + tmp_path, + "--expected-binding", + invalid, + "--max-observation-age-ms", + "1000", + raw=True, + ) + assert failure.returncode == 2 + assert "Traceback" not in failure.stderr + assert not failure.stdout + after = { + str(p.relative_to(tmp_path)): p.read_bytes() + for p in tmp_path.rglob("*") + if p.is_file() + } + assert after == before From 77aabe3297d1168d45651e92dc23a545f78c9a13 Mon Sep 17 00:00:00 2001 From: Dante Date: Mon, 26 Oct 2026 09:45:03 +0800 Subject: [PATCH 2/2] docs: define exact observer read eligibility and remaining qualification --- .../reliability_diagnostics/README.md | 19 +++++++++++++++++++ .../reliability_diagnostics/README.zh-CN.md | 16 ++++++++++++++++ 2 files changed, 35 insertions(+) diff --git a/loopx/capabilities/reliability_diagnostics/README.md b/loopx/capabilities/reliability_diagnostics/README.md index d3b8560af4..8a52e10343 100644 --- a/loopx/capabilities/reliability_diagnostics/README.md +++ b/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 ''` +with `--as-of ` and `--max-observation-age-ms `. +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 diff --git a/loopx/capabilities/reliability_diagnostics/README.zh-CN.md b/loopx/capabilities/reliability_diagnostics/README.zh-CN.md index 8dc69a8221..e02e84fd10 100644 --- a/loopx/capabilities/reliability_diagnostics/README.zh-CN.md +++ b/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 ''`、显式 +`--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 或启用轮询。