Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -532,8 +532,14 @@ 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.
overhead measurement, and deployment-owner acceptance of the ledger retention
and deletion profile from decision 4 below. The
[local retention reference (v0)](../../../loopx/capabilities/reliability_diagnostics/docs/local-retention-v0.md)
and its synthetic CLI lifecycle smoke supply an offline export/delete/restore
rehearsal, including installed-package readback and preservation of negative
evidence. They do not implement automated retention or prove real-observer
shutdown, filename/tenant isolation, C0/C1 or a deployment's deletion policy.
Implementation and milestone evidence remain in [task #5211](https://github.com/loopx-project/loopx/issues/5211).

### P1 — Benchmark-qualified diagnostic pilot

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -453,7 +453,13 @@ overhead;不存在 production authority。
`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。
observer run、overhead 测量报告,以及部署 owner 对下文 decision 4 的 ledger retention 与
deletion profile 的接受决定。
[本地 retention 参考方案(v0)](../../../loopx/capabilities/reliability_diagnostics/docs/local-retention-v0.md#中文)
及其合成 CLI lifecycle smoke 提供离线导出/删除/恢复演练,包含安装包读回与负面证据保留;
它们没有实现自动 retention,也未证明真实 observer 停写、文件名/租户隔离、C0/C1 或部署的
删除 policy。实现 PR 和 milestone 证据继续归入
[task #5211](https://github.com/loopx-project/loopx/issues/5211)。

### P1 — Benchmark-qualified diagnostic pilot

Expand Down
151 changes: 151 additions & 0 deletions examples/reliability_diagnostics/ledger-retention-smoke.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,151 @@
#!/usr/bin/env python3
"""Rehearse offline diagnostic export/delete/restore in disposable state only.

The invariant is byte-preserving recovery of negative evidence through the
real CLI. This is neither a live observer run nor automatic retention policy.
Use --installed with a non-editable installed package to check its entrypoint.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import subprocess
import sys
import tempfile
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parents[2]
AS_OF = "2026-09-01T12:00:00+00:00"


def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--installed", action="store_true")
args = parser.parse_args()
if not args.installed:
sys.path.insert(0, str(REPO_ROOT))

import loopx
from loopx.capabilities.reliability_diagnostics import (
FIXTURE_GOAL_ID,
run_dsh_fixture,
)

if args.installed:
assert not Path(loopx.__file__).resolve().is_relative_to(REPO_ROOT), (
"--installed requires a non-editable installation outside the checkout"
)

fixture = run_dsh_fixture()
data = "".join(json.dumps(row) + "\n" for row in fixture["ledger_records"])
env = dict(os.environ)
env.pop("PYTHONPATH", None)
if not args.installed:
env["PYTHONPATH"] = str(REPO_ROOT)

with tempfile.TemporaryDirectory(prefix="loopx-retention-smoke-") as tmp:
root = Path(tmp)
runtime = root / "runtime"
runtime.mkdir()
# Synthetic siblings must survive operations on the one ledger.
siblings = {
runtime / name: b'{"synthetic":"unchanged"}\n'
for name in ("goal.json", "todo.json", "quota.json", "gate.json", "session.json")
}
for path, content in siblings.items():
path.write_bytes(content)

def cli(command: str, *options: str, at: Path = runtime, stdin: str | None = None) -> dict:
result = subprocess.run(
[sys.executable, "-m", "loopx.cli", "--runtime-root", str(at),
"--format", "json", "reliability-diagnostics", command,
"--goal-id", FIXTURE_GOAL_ID, *options],
cwd=root, env=env, input=stdin, capture_output=True, text=True,
check=False, timeout=30,
)
assert result.returncode == 0, result.stderr or result.stdout
payload = json.loads(result.stdout)
assert payload["ok"] is True, payload
return payload

def readback(at: Path = runtime) -> dict:
return cli("status", "--with-receipt", "--as-of", AS_OF, at=at)

ingest = cli("ingest", "--input", "-", stdin=data)
ledger = runtime / ingest["ledger_ref"]
assert ingest["rejected_event_count"] == 0

for expected_status in ("degraded", "invalid"):
if expected_status == "invalid":
# A refused control field leaves a durable marker. Recovery must
# preserve it instead of selecting only accepted envelopes.
refused = dict(fixture["ledger_records"][0], command={"kind": "stop"})
refusal = cli("ingest", "--input", "-", stdin=json.dumps(refused) + "\n")
assert refusal["ingest_gate_recorded"] is True
assert refusal["rejected_by_reason"] == {"control_field_rejected": 1}
assert b"reliability_ingest_violation_v0" in ledger.read_bytes()
assert b'"command"' not in ledger.read_bytes()

before = readback()
receipt = before["receipt"]
assert receipt["status"] == expected_status, receipt
assert {"sequence_gap", "backpressure_drop", "raw_material_rejected",
"clock_uncertainty_exceeded"} <= set(receipt["reason_codes"])
assert receipt["lost_event_count"] == 2
assert receipt["backpressure_drop_count"] == 3
assert receipt["clock"]["max_uncertainty_ms"] == 1500

content = ledger.read_bytes()
digest = hashlib.sha256(content).hexdigest()
archive = root / f"archive-{expected_status}"
copy_runtime = archive / "readback-runtime"
copy_ledger = copy_runtime / ingest["ledger_ref"]
copy_ledger.parent.mkdir(parents=True)
copy_ledger.write_bytes(content)
(archive / "readback.json").write_text(json.dumps(before), encoding="utf-8")
(archive / "ledger.sha256").write_text(digest, encoding="ascii")
assert hashlib.sha256(copy_ledger.read_bytes()).hexdigest() == digest
assert copy_ledger.read_bytes() == ledger.read_bytes()
assert readback(copy_runtime) == before

# Tampering with a private export cannot be treated as verified.
damaged = content + b"{}\n"
assert hashlib.sha256(damaged).hexdigest() != digest
assert hashlib.sha256(ledger.read_bytes()).hexdigest() == digest
ledger.unlink()
missing = readback()
assert missing["receipt"]["status"] == "invalid"
assert "no_observations" in missing["receipt"]["reason_codes"]
assert missing["receipt"]["persisted_event_count"] == 0

# Exclusive creation is the restore guard: never merge or overwrite
# a file created by a restarted observer.
with ledger.open("xb") as handle:
handle.write(copy_ledger.read_bytes())
try:
ledger.open("xb").close()
except FileExistsError:
pass
else:
raise AssertionError("restore must refuse an existing ledger")
assert hashlib.sha256(ledger.read_bytes()).hexdigest() == digest
assert readback() == before
assert all(path.read_bytes() == value for path, value in siblings.items())
projection = before["projection"]
assert projection["mode"] == "read_only"
assert projection["authority"] == "none"
assert projection["worker_influence"] == "none"
assert receipt["outbound_endpoints"] == []
assert receipt["observation_entered_worker_context"] is False
assert receipt["observation_entered_scheduler_inputs"] is False

print("reliability-diagnostics ledger-retention-smoke: ok "
f"(synthetic; installed={args.installed}; degraded/invalid evidence preserved)")
return 0


if __name__ == "__main__":
raise SystemExit(main())
8 changes: 8 additions & 0 deletions loopx/capabilities/reliability_diagnostics/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -193,10 +193,18 @@ never be silently attributed to the configured goal. Token-level
`assistant/chunk` events — a retired type that only older durable logs still
replay — are not consumed.

For an operator-run offline export/delete/restore rehearsal, use the
[local retention reference (v0)](docs/local-retention-v0.md). It preserves whole
ledger bytes and negative integrity evidence, requires a frozen writer and an
owner-selected finite retention period, and does not implement automatic TTL
or qualify a live deployment. It also documents the current filename-alias
boundary; ambiguous ownership must hold deletion.

## Validation

```bash
python3 examples/reliability_diagnostics/dsh-shadow-observer-fixture-smoke.py
uv run --extra test python examples/reliability_diagnostics/ledger-retention-smoke.py
python3 -m pytest tests/capabilities/test_reliability_diagnostics.py tests/capabilities/test_reliability_diagnostics_dsh_provider.py -q
cd packages/dsh-loopx-plugin && pnpm typecheck && pnpm test -- observer
```
Expand Down
6 changes: 6 additions & 0 deletions loopx/capabilities/reliability_diagnostics/README.zh-CN.md
Original file line number Diff line number Diff line change
Expand Up @@ -165,10 +165,16 @@ parity)。启用后,`observer.ts` 只观察 `session/created`、`session/eve
到配置的 goal。token 级 `assistant/chunk`(已退场类型,只有旧 durable 日志还会重放)
不被消费。

人工离线导出/删除/恢复演练使用
[本地 retention 参考方案(v0)](docs/local-retention-v0.md#中文)。它保留完整 ledger 字节与
负面完整性证据,要求冻结 writer 并由 owner 选择有限保留期限;没有实现自动 TTL,也不构成
live 部署验收。方案说明了当前文件名别名边界,归属不明确时须暂停删除。

## 验证

```bash
python3 examples/reliability_diagnostics/dsh-shadow-observer-fixture-smoke.py
uv run --extra test python examples/reliability_diagnostics/ledger-retention-smoke.py
python3 -m pytest tests/capabilities/test_reliability_diagnostics.py tests/capabilities/test_reliability_diagnostics_dsh_provider.py -q
cd packages/dsh-loopx-plugin && pnpm typecheck && pnpm test -- observer
```
Expand Down
Loading
Loading