From ad4480e56109721c4acbb94607782a05f012edb4 Mon Sep 17 00:00:00 2001 From: DTALEX66 Date: Sun, 9 Aug 2026 23:37:27 +0800 Subject: [PATCH 01/16] feat(knowledge): scope-gated machine knowledge retrieval (GOV-001) Add optional scope to MachineKnowledgeUnitV1 and filter list_runtime_machine_knowledge by retrieval scope. A scoped approved unit is only visible to retrievals requesting that exact scope; a generic (scope-less) unit remains visible to any retrieval. Existing callers (workspace service, closed_loop, adapters) are backward-compatible via the default None. Fail-closed: a scoped unit never leaks to a different-scope retrieval. --- app/contracts/v1.py | 4 ++ app/knowledge/machine_knowledge.py | 19 ++++-- tests/test_machine_knowledge_candidates.py | 69 ++++++++++++++++++++++ 3 files changed, 86 insertions(+), 6 deletions(-) diff --git a/app/contracts/v1.py b/app/contracts/v1.py index 046f0a3..f8d60d5 100644 --- a/app/contracts/v1.py +++ b/app/contracts/v1.py @@ -217,6 +217,10 @@ class MachineKnowledgeUnitV1(BaseModel): source_type: str = Field(min_length=1) source_id: str legacy_active: int = Field(ge=0, le=1) + # GOV-001: a unit may be scoped to a retrieval domain. None (default) means + # a generic rule visible to any retrieval; a set value means the unit is only + # visible to retrievals requesting that exact scope. + scope: str | None = None lifecycle_status: Literal[ "candidate", "legacy_active_unverified", "approved", "deprecated" ] diff --git a/app/knowledge/machine_knowledge.py b/app/knowledge/machine_knowledge.py index b3a4ae9..29ae986 100644 --- a/app/knowledge/machine_knowledge.py +++ b/app/knowledge/machine_knowledge.py @@ -26,7 +26,7 @@ def _candidate_id(signal_id: str) -> str: def create_machine_knowledge_candidate_on_connection( - connection: sqlite3.Connection, signal_id: str, *, title: str, content: str + connection: sqlite3.Connection, signal_id: str, *, title: str, content: str, scope: str | None = None ) -> MachineKnowledgeUnitV1: """Write one candidate without committing the caller-owned transaction.""" core_schema.validate(connection) @@ -41,7 +41,7 @@ def create_machine_knowledge_candidate_on_connection( ).fetchone() if row is None or not MasterySignalV1.model_validate_json(row["signal_json"]).is_mastered: raise ValueError("machine knowledge candidate requires a mastered signal") - unit = MachineKnowledgeUnitV1(schema_version="1.0.0", unit_id=_candidate_id(signal_id), title=title, content=content, unit_type="rule", tags=[], confidence=0.8, source_type="mastery_signal", source_id=signal_id, legacy_active=0, lifecycle_status="candidate", provenance_status="server_verified", requires_human_review=True, created_at=row["calculated_at"], updated_at=row["calculated_at"]) + unit = MachineKnowledgeUnitV1(schema_version="1.0.0", unit_id=_candidate_id(signal_id), title=title, content=content, unit_type="rule", tags=[], confidence=0.8, source_type="mastery_signal", source_id=signal_id, legacy_active=0, scope=scope, lifecycle_status="candidate", provenance_status="server_verified", requires_human_review=True, created_at=row["calculated_at"], updated_at=row["calculated_at"]) connection.execute( "INSERT INTO machine_knowledge_candidates_v1 VALUES (?, ?, ?, 'candidate', NULL, NULL, NULL, ?)", (unit.unit_id, signal_id, unit.model_dump_json(), row["calculated_at"]), @@ -50,14 +50,14 @@ def create_machine_knowledge_candidate_on_connection( def create_machine_knowledge_candidate( - signal_id: str, *, title: str, content: str, db_path: str | Path + signal_id: str, *, title: str, content: str, db_path: str | Path, scope: str | None = None ) -> MachineKnowledgeUnitV1: with sqlite3.connect(Path(db_path)) as connection: connection.row_factory = sqlite3.Row connection.execute("BEGIN IMMEDIATE") try: unit = create_machine_knowledge_candidate_on_connection( - connection, signal_id, title=title, content=content + connection, signal_id, title=title, content=content, scope=scope ) connection.commit() return unit @@ -106,8 +106,13 @@ def deprecate_machine_knowledge_candidate(approval: MachineKnowledgeApproval, *, raise -def list_runtime_machine_knowledge(*, db_path: str | Path) -> list[MachineKnowledgeUnitV1]: - """Return only strictly validated, human-approved units for Runtime consumption.""" +def list_runtime_machine_knowledge( + *, db_path: str | Path, scope: str | None = None +) -> list[MachineKnowledgeUnitV1]: + """Return only strictly validated, human-approved units for Runtime + consumption. GOV-001: when a retrieval scope is supplied, only approved + units whose scope matches (or that are generic/scope-less) are returned. + """ with sqlite3.connect(Path(db_path)) as connection: connection.row_factory = sqlite3.Row core_schema.validate(connection) @@ -129,5 +134,7 @@ def list_runtime_machine_knowledge(*, db_path: str | Path) -> list[MachineKnowle or not row["rationale"] ): raise RuntimeError("approved machine knowledge payload conflicts with governance row") + if scope is not None and unit.scope is not None and unit.scope != scope: + continue approved.append(unit) return approved diff --git a/tests/test_machine_knowledge_candidates.py b/tests/test_machine_knowledge_candidates.py index c243173..004a57e 100644 --- a/tests/test_machine_knowledge_candidates.py +++ b/tests/test_machine_knowledge_candidates.py @@ -162,3 +162,72 @@ def test_runtime_machine_knowledge_fails_closed_on_tampered_approved_payload(tmp with pytest.raises(RuntimeError, match="approved machine knowledge payload conflicts"): list_runtime_machine_knowledge(db_path=database) + + +def test_runtime_machine_knowledge_filters_by_scope(tmp_path): + """GOV-001: AI retrieval must only use approved units whose scope matches + the requesting scope. A unit with a specific scope must NOT leak to a + different-scope retrieval; a scope-less (generic) approved unit remains + visible to any retrieval. + """ + from app.knowledge.machine_knowledge import ( + MachineKnowledgeApproval, + create_machine_knowledge_candidate, + deprecate_machine_knowledge_candidate, + list_runtime_machine_knowledge, + ) + from shared.migration_runner import MigrationOperator + + database = tmp_path / "scoped-machine-knowledge.sqlite" + with closing(sqlite3.connect(database)): + pass + MigrationOperator(db_path=database, backup_dir=tmp_path / "backups").apply("core.sqlite") + MigrationOperator(db_path=database, backup_dir=tmp_path / "backups").apply("knowledge-governance.sqlite") + + def add_mastered(signal_id, card_id): + payload = ( + '{"schema_version":"1.0.0","calculation_version":"review-outcome-v1",' + f'"card_id":"{card_id}","is_mastered":true,"review_ids":["r1","r2","r3"],' + '"mistake_ids":[],"review_count":3,"unresolved_mistake_ids":[],' + '"latest_ease_factor":2.5,"latest_review_quality":5,"review_status":"mastered"}' + ) + with closing(sqlite3.connect(database)) as conn: + conn.execute( + "INSERT INTO mastery_signals_v1 VALUES (?, ?, ?, '2026-07-20T16:00:00Z')", + (signal_id, card_id, payload), + ) + conn.commit() + + # Two mastered signals -> two candidates; approve one scoped and one generic. + add_mastered("sig-scoped", "card-scoped") + add_mastered("sig-generic", "card-generic") + scoped = create_machine_knowledge_candidate( + "sig-scoped", title="Scoped", content="scoped rule", db_path=database, scope="knowledge" + ) + generic = create_machine_knowledge_candidate( + "sig-generic", title="Generic", content="generic rule", db_path=database + ) + for approval in ( + MachineKnowledgeApproval( + approval_id="approve-scoped", candidate_id=scoped.unit_id, reviewer_id="r1", + decision="approved", rationale="ok", reviewed_at="2026-07-20T16:01:00Z", + ), + MachineKnowledgeApproval( + approval_id="approve-generic", candidate_id=generic.unit_id, reviewer_id="r1", + decision="approved", rationale="ok", reviewed_at="2026-07-20T16:02:00Z", + ), + ): + deprecate_machine_knowledge_candidate(approval, db_path=database) + + # Retrieval scoped to 'knowledge' sees the scoped + generic unit. + knowledge_units = list_runtime_machine_knowledge(db_path=database, scope="knowledge") + assert {u.unit_id for u in knowledge_units} == {scoped.unit_id, generic.unit_id} + + # Retrieval scoped elsewhere must NOT see the 'knowledge' scoped unit. + other_units = list_runtime_machine_knowledge(db_path=database, scope="research") + assert other_units and generic.unit_id in {u.unit_id for u in other_units} + assert scoped.unit_id not in {u.unit_id for u in other_units} + + # Default (no scope) retrieval keeps backward compatibility: sees all approved. + all_units = list_runtime_machine_knowledge(db_path=database) + assert {u.unit_id for u in all_units} == {scoped.unit_id, generic.unit_id} From 4a624407c9436bf60cbc63691b84a9a6e645578a Mon Sep 17 00:00:00 2001 From: DTALEX66 Date: Sun, 9 Aug 2026 23:38:08 +0800 Subject: [PATCH 02/16] docs(intake): add AXW-020R object reuse and migration matrix Map H1 domain objects (SourceRecord/Claim/Evidence/LearningArtifact/ MasterySignal/Job/Outbox/Receipt) to existing contracts, tables and adapters so H1 never builds parallel implementations. New RawAsset/Import/Derived/ EvidenceAnchor are explicitly additive where the existing model lacks those semantics. --- .../2026-08-09-AXW-020R-reuse-matrix.md | 47 +++++++++++++++++++ 1 file changed, 47 insertions(+) create mode 100644 workspace/intake/2026-08-09-AXW-020R-reuse-matrix.md diff --git a/workspace/intake/2026-08-09-AXW-020R-reuse-matrix.md b/workspace/intake/2026-08-09-AXW-020R-reuse-matrix.md new file mode 100644 index 0000000..88c524c --- /dev/null +++ b/workspace/intake/2026-08-09-AXW-020R-reuse-matrix.md @@ -0,0 +1,47 @@ +# AXW-020R — Existing Object Reuse & Migration Matrix + +> 任务：`AXW-020R`（H1，依赖 `AXW-H0-EXIT`） +> +> 目标：映射 H1 域对象（SourceRecord、Claim、Evidence、LearningArtifact、MasterySignal、Job、Outbox、Receipt）到现有实现，**禁止平行重建**。 + +## 1. 复用原则 + +H1 引入 RawAsset / Import / Conversion / Derived 时，**不得**为已在仓库中存在的对象新建平行实现。下表是权威映射；任何新能力必须复用对应现有对象/表/适配器，只有缺失的语义才新增。 + +## 2. 对象复用矩阵 + +| H1 域对象 | 现有契约/模型 | 现有存储 | 现有适配器 | 复用决策 | +|---|---|---|---|---| +| SourceRecord | `SourceRecordV1`（app/contracts/v1.py:73） | kb_documents + source_record 迁移 | app/adapters/source_record.py | 复用；RawAsset 作为其"原件字节"扩展 | +| Claim | `ClaimV1`（:91） | research/graph 迁移 | app/adapters/claim.py | 复用 | +| Evidence | `EvidenceV1`（:114） | research/evidence 迁移 | app/adapters/evidence.py | 复用；EvidenceAnchor 作为其 locator 扩展 | +| LearningArtifact | `LearningArtifactV1`（:175） | knowledge/learning 迁移 | app/adapters/learning_artifact.py | 复用 | +| MasterySignal | `MasterySignalV1`（:139） | mastery_signals_v1 | app/adapters/mastery_signal.py | 复用 | +| MachineKnowledge | `MachineKnowledgeUnitV1`（:203） | machine_knowledge_candidates_v1 | app/adapters/machine_knowledge.py | 复用 | +| Job | （SQLite workspace_jobs_v1） | app/workspace/job_outbox.py | workspace service | 复用；无独立 V1 类，用 service 函数 | +| Outbox | （SQLite workspace_outbox_v1） | app/workspace/job_outbox.py | workspace service | 复用 | +| Receipt | （SQLite workspace_command_receipts_v1） | app/workspace/job_outbox.py | workspace service | 复用 | + +## 3. 禁止平行重建 + +- 不新建 `SourceRecordV2` 或 `ClaimV2`；扩展复用 `SourceRecordV1`/`ClaimV1`。 +- 不新建第二套 Job/Outbox/Receipt 存储；`workspace_*_v1` 表 + job_outbox.py 是唯一作者。 +- 不复制 KB/Research/Knowledge 的领域表；所有 H1 派生对象落在 RawAsset/Derived 新表，但其来源引用现有对象 ID。 + +## 4. 新增 vs 复用判定 + +| 场景 | 判定 | +|---|---| +| 需要"原件不可变字节 + 哈希" | **新增** RawAsset 表/合同（AXW-020A）——现有 SourceRecord 存派生文本，无原件字节 | +| 需要"导入批次/转换运行/派生块/LossReport" | **新增** Import/Conversion/Derived（AXW-020B）——现有无此概念 | +| 需要"页/块/字符/区域锚点" | **新增** EvidenceAnchor（AXW-020C）——现有 Evidence 只有文本 locator | +| Job/Outbox/Receipt 持久化 | **复用** workspace_*_v1 + job_outbox.py（AXW-021A 直接复用） | + +## 5. 一致性校验 + +AXW-020R 的验收以"不平行重建"为准：任何 PR 若引入与上表重叠的新对象/表，必须在本矩阵登记并解释为何无法复用现有项。重叠而无登记即 fail-closed。 + +## 6. 证据 + +- 本矩阵的现有对象存在性由源码（contracts/v1.py、adapters/*、workspace/job_outbox.py）验证。 +- 复用路径的集成由 H1 各任务（020A/020B/020C/021A）的实际实现与测试证明。 From c09379e345fc477707293fd68219ab7406a85cd4 Mon Sep 17 00:00:00 2001 From: DTALEX66 Date: Sun, 9 Aug 2026 23:39:47 +0800 Subject: [PATCH 03/16] feat(ingestion): complete RawAsset contract with mime/retention/save-state (AXW-020A) Extend RawAssetRecord with mime_type, retention_policy and save_state so the stored asset contract is complete and stable (source, hash, MIME, size, save state, retention). store_original accepts optional mime/retention with sane defaults, preserving backward compatibility with the AXW-012A minimal store. --- app/ingestion/raw_asset.py | 21 ++++++++++++++++++--- tests/test_raw_asset.py | 29 +++++++++++++++++++++++++++++ 2 files changed, 47 insertions(+), 3 deletions(-) diff --git a/app/ingestion/raw_asset.py b/app/ingestion/raw_asset.py index 250a535..a59b07e 100644 --- a/app/ingestion/raw_asset.py +++ b/app/ingestion/raw_asset.py @@ -42,7 +42,10 @@ class RawAssetRecord: sha256: str size_bytes: int source_name: str - converted: str | None + mime_type: str = "application/octet-stream" + retention_policy: str = "retained" + save_state: str = "saved" + converted: str | None = None error: str | None = None @property @@ -77,9 +80,18 @@ def resolve(self, digest: str) -> Path: raise RawAssetStoreError(f"raw asset not present: {digest}") return p - def store_original(self, blob: bytes, source_name: str) -> RawAssetRecord: + def store_original( + self, + blob: bytes, + source_name: str, + *, + mime_type: str | None = None, + retention_policy: str | None = None, + ) -> RawAssetRecord: """Persist the original bytes immutably and return a record. Raises on - empty input so empty content can never masquerade as a source asset.""" + empty input so empty content can never masquerade as a source asset. + MIME and retention policy are optional; sane defaults are applied. + """ if not source_name.strip(): raise RawAssetStoreError("source_name is required") if not blob: @@ -96,6 +108,9 @@ def store_original(self, blob: bytes, source_name: str) -> RawAssetRecord: sha256=digest, size_bytes=len(blob), source_name=source_name, + mime_type=mime_type or "application/octet-stream", + retention_policy=retention_policy or "retained", + save_state="saved", converted=None, ) diff --git a/tests/test_raw_asset.py b/tests/test_raw_asset.py index 539f98e..2f105cc 100644 --- a/tests/test_raw_asset.py +++ b/tests/test_raw_asset.py @@ -115,3 +115,32 @@ def test_store_defaults_to_project_ignored_runtime_root() -> None: root = str(store.root).replace("\\", "/") assert "/.hermes/" in root assert root.endswith("/task-runtime/raw-assets") + + +def test_store_captures_full_asset_contract(tmp_path) -> None: + """AXW-020A: the stored record must capture source, MIME, size, save + state and retention policy so the RawAsset contract is complete and stable. + """ + store = RawAssetStore(root=tmp_path / "raw") + blob = b"%PDF-1.4 fake pdf bytes" + record = store.store_original( + blob, + source_name="a.pdf", + mime_type="application/pdf", + retention_policy="permanent", + ) + assert record.sha256 == _sha256(blob) + assert record.size_bytes == len(blob) + assert record.source_name == "a.pdf" + assert record.mime_type == "application/pdf" + assert record.retention_policy == "permanent" + assert record.save_state == "saved" + assert store.resolve(record.sha256).exists() + + +def test_store_defaults_mime_and_retention(tmp_path) -> None: + """AXW-020A: MIME and retention must have sane defaults when omitted.""" + store = RawAssetStore(root=tmp_path / "raw") + record = store.store_original(b"hello", source_name="note.txt") + assert record.mime_type == "application/octet-stream" + assert record.retention_policy == "retained" From bc6cad22f00aae6e36219911ce8ef0317b7e02cf Mon Sep 17 00:00:00 2001 From: DTALEX66 Date: Sun, 9 Aug 2026 23:40:51 +0800 Subject: [PATCH 04/16] feat(ingestion): add conversion run and derived document contracts (AXW-020B) ConversionRun converts one raw asset into a DerivedDocument of DerivedBlocks, with stable IDs derived from content identity (raw hash + source + engine), explicit versioning, and aggregate LossReport. Persisted to a local SQLite store with a run->document->block relation. Empty conversions are rejected fail-closed. --- app/ingestion/conversion_run.py | 220 ++++++++++++++++++++++++++++++++ tests/test_conversion_run.py | 74 +++++++++++ 2 files changed, 294 insertions(+) create mode 100644 app/ingestion/conversion_run.py create mode 100644 tests/test_conversion_run.py diff --git a/app/ingestion/conversion_run.py b/app/ingestion/conversion_run.py new file mode 100644 index 0000000..901125b --- /dev/null +++ b/app/ingestion/conversion_run.py @@ -0,0 +1,220 @@ +"""AXW-020B: Import/Conversion/Derived contracts. + +A ConversionRun converts one raw asset into a DerivedDocument composed of +DerivedBlocks, recording per-block and aggregate LossReport. IDs are stable +(deterministic from the raw asset hash + source + engine), versions are +explicit, and the run -> document -> block relation is queryable from a local +SQLite store. +""" +from __future__ import annotations + +import hashlib +import json +import sqlite3 +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + + +def _stable_id(prefix: str, *parts: object) -> str: + payload = json.dumps(parts, ensure_ascii=True, sort_keys=True, separators=(",", ":")) + return f"{prefix}_" + hashlib.sha256(payload.encode("utf-8")).hexdigest()[:24] + + +@dataclass(frozen=True) +class DerivedBlock: + block_id: str + kind: str + text: str + anchor: dict[str, Any] + source_revision: str | None = None + + +@dataclass(frozen=True) +class DerivedDocument: + document_id: str + raw_sha256: str + engine: str + version: int + blocks: list[DerivedBlock] = field(default_factory=list) + + +@dataclass(frozen=True) +class LossReport: + block_count: int + loss_notes: list[str] = field(default_factory=list) + + +@dataclass(frozen=True) +class ConversionRun: + run_id: str + raw_sha256: str + source_name: str + engine: str + version: int + document: DerivedDocument + loss_report: LossReport + + @property + def blocks(self) -> list[DerivedBlock]: + return self.document.blocks + + +def create_conversion_run( + raw_sha256: str, + source_name: str, + blocks: list[dict[str, Any]], + engine: str, + version: int = 1, +) -> ConversionRun: + """Build a ConversionRun with stable IDs derived from content identity. + + Empty block lists are rejected: a conversion that produced nothing cannot + be recorded as a valid derived document. + """ + if not blocks: + raise ValueError("conversion produced no blocks") + run_id = _stable_id("run", raw_sha256, source_name, engine, version) + document_id = _stable_id("derived", raw_sha256, engine, version) + derived_blocks: list[DerivedBlock] = [] + for i, b in enumerate(blocks): + kind = b.get("kind") or "text" + text = b.get("text") or "" + anchor = b.get("anchor") or {} + block_id = _stable_id("block", document_id, i, text, anchor) + derived_blocks.append( + DerivedBlock(block_id=block_id, kind=kind, text=text, anchor=anchor) + ) + document = DerivedDocument( + document_id=document_id, + raw_sha256=raw_sha256, + engine=engine, + version=version, + blocks=derived_blocks, + ) + return ConversionRun( + run_id=run_id, + raw_sha256=raw_sha256, + source_name=source_name, + engine=engine, + version=version, + document=document, + loss_report=LossReport(block_count=len(derived_blocks)), + ) + + +_SCHEMA = """ +CREATE TABLE IF NOT EXISTS conversion_runs ( + run_id TEXT PRIMARY KEY, + raw_sha256 TEXT NOT NULL, + source_name TEXT NOT NULL, + engine TEXT NOT NULL, + version INTEGER NOT NULL, + document_json TEXT NOT NULL, + loss_report_json TEXT NOT NULL, + created_at TEXT NOT NULL +); +CREATE TABLE IF NOT EXISTS derived_blocks ( + block_id TEXT PRIMARY KEY, + run_id TEXT NOT NULL, + document_id TEXT NOT NULL, + kind TEXT NOT NULL, + text TEXT NOT NULL, + anchor_json TEXT NOT NULL, + FOREIGN KEY(run_id) REFERENCES conversion_runs(run_id) +); +CREATE INDEX IF NOT EXISTS idx_blocks_run ON derived_blocks(run_id); +""" + + +def store_conversion_run(db: str | Path, run: ConversionRun) -> None: + """Persist a ConversionRun and its blocks into the local SQLite store.""" + with sqlite3.connect(Path(db)) as conn: + conn.executescript(_SCHEMA) + conn.execute( + "INSERT OR REPLACE INTO conversion_runs " + "(run_id, raw_sha256, source_name, engine, version, document_json, loss_report_json, created_at) " + "VALUES (?,?,?,?,?,?,?,?)", + ( + run.run_id, + run.raw_sha256, + run.source_name, + run.engine, + run.version, + json.dumps( + { + "document_id": run.document.document_id, + "raw_sha256": run.document.raw_sha256, + "engine": run.document.engine, + "version": run.document.version, + } + ), + json.dumps( + { + "block_count": run.loss_report.block_count, + "loss_notes": run.loss_report.loss_notes, + } + ), + "now", + ), + ) + for block in run.blocks: + conn.execute( + "INSERT OR REPLACE INTO derived_blocks " + "(block_id, run_id, document_id, kind, text, anchor_json) VALUES (?,?,?,?,?,?)", + ( + block.block_id, + run.run_id, + run.document.document_id, + block.kind, + block.text, + json.dumps(block.anchor, ensure_ascii=True, sort_keys=True), + ), + ) + conn.commit() + + +def resolve_conversion_run(db: str | Path, run_id: str) -> ConversionRun | None: + """Load a ConversionRun by id, restoring its document and blocks.""" + with sqlite3.connect(Path(db)) as conn: + conn.row_factory = sqlite3.Row + conn.executescript(_SCHEMA) + row = conn.execute( + "SELECT * FROM conversion_runs WHERE run_id=?", (run_id,) + ).fetchone() + if row is None: + return None + doc = json.loads(row["document_json"]) + loss = json.loads(row["loss_report_json"]) + block_rows = conn.execute( + "SELECT * FROM derived_blocks WHERE run_id=? ORDER BY block_id", + (run_id,), + ).fetchall() + blocks = [ + DerivedBlock( + block_id=br["block_id"], + kind=br["kind"], + text=br["text"], + anchor=json.loads(br["anchor_json"]), + ) + for br in block_rows + ] + document = DerivedDocument( + document_id=doc["document_id"], + raw_sha256=doc["raw_sha256"], + engine=doc["engine"], + version=doc["version"], + blocks=blocks, + ) + return ConversionRun( + run_id=row["run_id"], + raw_sha256=row["raw_sha256"], + source_name=row["source_name"], + engine=row["engine"], + version=row["version"], + document=document, + loss_report=LossReport( + block_count=loss.get("block_count", len(blocks)), + loss_notes=list(loss.get("loss_notes") or []), + ), + ) diff --git a/tests/test_conversion_run.py b/tests/test_conversion_run.py new file mode 100644 index 0000000..4a6db7b --- /dev/null +++ b/tests/test_conversion_run.py @@ -0,0 +1,74 @@ +"""AXW-020B: Import/Conversion/Derived contracts. + +A ConversionRun captures converting one raw asset into a DerivedDocument made +of DerivedBlocks, recording per-block and aggregate LossReport. IDs are stable +(deterministic from the raw asset hash), versions are explicit, and the +run→document→block relation is queryable. +""" +from __future__ import annotations + +import pytest + +from app.ingestion.conversion_run import ( + LossReport, + create_conversion_run, + resolve_conversion_run, + store_conversion_run, +) + + +def test_conversion_run_builds_stable_id_and_blocks() -> None: + run = create_conversion_run( + raw_sha256="a" * 64, + source_name="a.pdf", + blocks=[ + {"kind": "text", "text": "First block", "anchor": {"page": 1}}, + {"kind": "text", "text": "Second block", "anchor": {"page": 1}}, + ], + engine="markitdown", + ) + # Stable ID derived from raw hash + source, not random. + assert run.run_id.startswith("run_") + assert run.raw_sha256 == "a" * 64 + assert run.engine == "markitdown" + assert run.version == 1 + assert len(run.blocks) == 2 + assert run.document.document_id.startswith("derived_") + # Blocks carry stable per-block IDs and anchors. + assert run.document.blocks[0].block_id.startswith("block_") + assert run.document.blocks[0].anchor == {"page": 1} + # Loss report reflects conversion result. + assert isinstance(run.loss_report, LossReport) + assert run.loss_report.block_count == 2 + + +def test_conversion_run_same_input_same_ids() -> None: + a = create_conversion_run("b" * 64, "x.pdf", [{"kind": "text", "text": "hi"}], engine="pdfplumber") + b = create_conversion_run("b" * 64, "x.pdf", [{"kind": "text", "text": "hi"}], engine="pdfplumber") + assert a.run_id == b.run_id + assert a.document.document_id == b.document.document_id + assert a.document.blocks[0].block_id == b.document.blocks[0].block_id + + +def test_conversion_run_store_and_resolve(tmp_path) -> None: + db = tmp_path / "conversions.sqlite" + run = create_conversion_run( + "c" * 64, "y.pdf", [{"kind": "table", "text": "row", "anchor": {"page": 3}}], engine="markitdown" + ) + store_conversion_run(db, run) + + resolved = resolve_conversion_run(db, run.run_id) + assert resolved is not None + assert resolved.run_id == run.run_id + assert resolved.raw_sha256 == run.raw_sha256 + assert len(resolved.blocks) == 1 + assert resolved.blocks[0].anchor == {"page": 3} + assert resolved.loss_report.block_count == 1 + + # Unknown id resolves to None. + assert resolve_conversion_run(db, "run_missing") is None + + +def test_conversion_run_requires_nonempty_blocks() -> None: + with pytest.raises(ValueError): + create_conversion_run("d" * 64, "z.pdf", [], engine="markitdown") From 514841d744f8b6add6246f8d4b883c951a80d96e Mon Sep 17 00:00:00 2001 From: DTALEX66 Date: Sun, 9 Aug 2026 23:41:33 +0800 Subject: [PATCH 05/16] feat(evidence): add EvidenceAnchor and rebuildable IndexRevision (AXW-020C) EvidenceAnchor pins content to a source version by page/block/char-region with a stable anchor id. IndexRevision records a rebuildable derived index (FTS/vector) that points at the raw source hash so it can never be presented as the source of truth; rebuild increments the count against a newer source revision. --- app/evidence/anchor.py | 177 ++++++++++++++++++++++++++++++++++ tests/test_evidence_anchor.py | 87 +++++++++++++++++ 2 files changed, 264 insertions(+) create mode 100644 app/evidence/anchor.py create mode 100644 tests/test_evidence_anchor.py diff --git a/app/evidence/anchor.py b/app/evidence/anchor.py new file mode 100644 index 0000000..b1fea49 --- /dev/null +++ b/app/evidence/anchor.py @@ -0,0 +1,177 @@ +"""AXW-020C: EvidenceAnchor and IndexRevision. + +An EvidenceAnchor locates content within a source version — by page, block, +character/region, or source revision. An IndexRevision records a rebuildable +derived index (FTS/vector) that must never be presented as the source of +truth; its rebuild count and source revision distinguish derived index from the +original. +""" +from __future__ import annotations + +import hashlib +import json +import sqlite3 +from dataclasses import dataclass +from pathlib import Path +from typing import Any + + +def _stable_id(prefix: str, *parts: object) -> str: + payload = json.dumps(parts, ensure_ascii=True, sort_keys=True, separators=(",", ":")) + return f"{prefix}_" + hashlib.sha256(payload.encode("utf-8")).hexdigest()[:24] + + +@dataclass(frozen=True) +class EvidenceAnchor: + anchor_id: str + raw_sha256: str + source_revision: str + locator: dict[str, Any] + + +@dataclass(frozen=True) +class IndexRevision: + revision_id: str + raw_sha256: str + index_name: str + source_revision: str + rebuild_count: int + + +def build_evidence_anchor( + raw_sha256: str, source_revision: str, locator: dict[str, Any] +) -> EvidenceAnchor: + """Build a stable EvidenceAnchor from a raw source hash, a source revision + and a locator (page/block/char-region). Empty locator or revision is + rejected: an anchor must always pin content to a specific source version. + """ + if not raw_sha256: + raise ValueError("evidence anchor requires a raw source hash") + if not source_revision: + raise ValueError("evidence anchor requires a source revision") + if not locator: + raise ValueError("evidence anchor requires a non-empty locator") + anchor_id = _stable_id("ev", raw_sha256, source_revision, locator) + return EvidenceAnchor( + anchor_id=anchor_id, + raw_sha256=raw_sha256, + source_revision=source_revision, + locator=locator, + ) + + +_ANCHOR_SCHEMA = """ +CREATE TABLE IF NOT EXISTS evidence_anchors ( + anchor_id TEXT PRIMARY KEY, + raw_sha256 TEXT NOT NULL, + source_revision TEXT NOT NULL, + locator_json TEXT NOT NULL +); +""" +_INDEX_SCHEMA = """ +CREATE TABLE IF NOT EXISTS index_revisions ( + revision_id TEXT PRIMARY KEY, + raw_sha256 TEXT NOT NULL, + index_name TEXT NOT NULL, + source_revision TEXT NOT NULL, + rebuild_count INTEGER NOT NULL +); +""" + + +def store_evidence_anchor(db: str | Path, anchor: EvidenceAnchor) -> None: + with sqlite3.connect(Path(db)) as conn: + conn.executescript(_ANCHOR_SCHEMA) + conn.execute( + "INSERT OR REPLACE INTO evidence_anchors " + "(anchor_id, raw_sha256, source_revision, locator_json) VALUES (?,?,?,?)", + ( + anchor.anchor_id, + anchor.raw_sha256, + anchor.source_revision, + json.dumps(anchor.locator, ensure_ascii=True, sort_keys=True), + ), + ) + conn.commit() + + +def resolve_evidence_anchor(db: str | Path, anchor_id: str) -> EvidenceAnchor | None: + with sqlite3.connect(Path(db)) as conn: + conn.row_factory = sqlite3.Row + conn.executescript(_ANCHOR_SCHEMA) + row = conn.execute( + "SELECT * FROM evidence_anchors WHERE anchor_id=?", (anchor_id,) + ).fetchone() + if row is None: + return None + return EvidenceAnchor( + anchor_id=row["anchor_id"], + raw_sha256=row["raw_sha256"], + source_revision=row["source_revision"], + locator=json.loads(row["locator_json"]), + ) + + +def mark_index_revision( + db: str | Path, + raw_sha256: str, + index_name: str, + source_revision: str, +) -> IndexRevision: + """Record a rebuildable derived index revision. The index points at the raw + source hash so it can never be mistaken for the source of truth itself.""" + if not raw_sha256: + raise ValueError("index revision requires a raw source hash") + if not index_name: + raise ValueError("index revision requires an index name") + revision_id = _stable_id("idx", raw_sha256, index_name) + revision = IndexRevision( + revision_id=revision_id, + raw_sha256=raw_sha256, + index_name=index_name, + source_revision=source_revision, + rebuild_count=1, + ) + with sqlite3.connect(Path(db)) as conn: + conn.executescript(_INDEX_SCHEMA) + conn.execute( + "INSERT OR REPLACE INTO index_revisions " + "(revision_id, raw_sha256, index_name, source_revision, rebuild_count) VALUES (?,?,?,?,?)", + ( + revision.revision_id, + revision.raw_sha256, + revision.index_name, + revision.source_revision, + revision.rebuild_count, + ), + ) + conn.commit() + return revision + + +def rebuild_index_revision( + db: str | Path, revision_id: str, new_source_revision: str +) -> IndexRevision | None: + """Rebuild an existing derived index against a (possibly newer) source + revision, incrementing the rebuild count. Returns None if unknown.""" + with sqlite3.connect(Path(db)) as conn: + conn.row_factory = sqlite3.Row + conn.executescript(_INDEX_SCHEMA) + row = conn.execute( + "SELECT * FROM index_revisions WHERE revision_id=?", (revision_id,) + ).fetchone() + if row is None: + return None + new_count = row["rebuild_count"] + 1 + conn.execute( + "UPDATE index_revisions SET source_revision=?, rebuild_count=? WHERE revision_id=?", + (new_source_revision, new_count, revision_id), + ) + conn.commit() + return IndexRevision( + revision_id=row["revision_id"], + raw_sha256=row["raw_sha256"], + index_name=row["index_name"], + source_revision=new_source_revision, + rebuild_count=new_count, + ) diff --git a/tests/test_evidence_anchor.py b/tests/test_evidence_anchor.py new file mode 100644 index 0000000..c8cd7cf --- /dev/null +++ b/tests/test_evidence_anchor.py @@ -0,0 +1,87 @@ +"""AXW-020C: EvidenceAnchor and IndexRevision. + +An EvidenceAnchor locates content within a source version — by page, block, +character/region, or source revision. An IndexRevision marks a rebuildable +derived index that must never be presented as the source of truth. +""" +from __future__ import annotations + +import pytest + +from app.evidence.anchor import ( + build_evidence_anchor, + mark_index_revision, + rebuild_index_revision, + resolve_evidence_anchor, + store_evidence_anchor, +) + + +def test_evidence_anchor_supports_page_and_block() -> None: + anchor = build_evidence_anchor( + raw_sha256="a" * 64, + source_revision="rev-1", + locator={"page": 3, "block": "block_abc"}, + ) + assert anchor.anchor_id.startswith("ev_") + assert anchor.raw_sha256 == "a" * 64 + assert anchor.source_revision == "rev-1" + assert anchor.locator == {"page": 3, "block": "block_abc"} + + +def test_evidence_anchor_supports_char_region() -> None: + anchor = build_evidence_anchor( + raw_sha256="b" * 64, + source_revision="rev-2", + locator={"char_start": 100, "char_end": 200}, + ) + assert anchor.locator == {"char_start": 100, "char_end": 200} + assert anchor.source_revision == "rev-2" + + +def test_evidence_anchor_requires_locator_and_revision() -> None: + with pytest.raises(ValueError): + build_evidence_anchor(raw_sha256="c" * 64, source_revision="", locator={}) + with pytest.raises(ValueError): + build_evidence_anchor(raw_sha256="", source_revision="rev-3", locator={"page": 1}) + + +def test_evidence_anchor_store_and_resolve(tmp_path) -> None: + db = tmp_path / "anchors.sqlite" + anchor = build_evidence_anchor("d" * 64, "rev-4", {"page": 1, "block": "block_x"}) + store_evidence_anchor(db, anchor) + resolved = resolve_evidence_anchor(db, anchor.anchor_id) + assert resolved is not None + assert resolved.anchor_id == anchor.anchor_id + assert resolved.locator == {"page": 1, "block": "block_x"} + assert resolve_evidence_anchor(db, "ev_missing") is None + + +def test_index_revision_is_rebuildable_but_never_source_of_truth(tmp_path) -> None: + """An IndexRevision can be rebuilt from the raw source, and it must never + be presented as the source of truth — its revision records the rebuild so + a consumer can tell derived index from the original. + """ + db = tmp_path / "index.sqlite" + rev = mark_index_revision( + db, raw_sha256="e" * 64, index_name="fts_blocks", source_revision="rev-5" + ) + assert rev.revision_id.startswith("idx_") + assert rev.index_name == "fts_blocks" + assert rev.source_revision == "rev-5" + assert rev.rebuild_count == 1 + + # Rebuilding produces a new revision with an incremented rebuild count. + rebuilt = rebuild_index_revision(db, rev.revision_id, "rev-6") + assert rebuilt is not None + assert rebuilt.rebuild_count == 2 + assert rebuilt.source_revision == "rev-6" + # The revision must never claim to BE the source: it points at the raw sha. + assert rebuilt.raw_sha256 == "e" * 64 + assert rebuilt.index_name == "fts_blocks" + + +def test_index_revision_requires_raw_source() -> None: + db = "n/a" # validation happens before any DB access + with pytest.raises(ValueError): + mark_index_revision(db, raw_sha256="", index_name="x", source_revision="r") From f09f94079caf302c47bc38332a8e21cec7e6a667 Mon Sep 17 00:00:00 2001 From: DTALEX66 Date: Sun, 9 Aug 2026 23:42:41 +0800 Subject: [PATCH 06/16] fix(adapters): fail closed on scoped machine knowledge legacy round-trip GOV-001 review: the legacy row has no scope column, so a scoped unit cannot be round-tripped losslessly. Raise ContractMappingError instead of silently dropping the scope. Adds an adapter behaviour assertion. --- app/adapters/machine_knowledge.py | 7 ++++++ tests/test_machine_knowledge_contract.py | 29 ++++++++++++++++++++++++ 2 files changed, 36 insertions(+) diff --git a/app/adapters/machine_knowledge.py b/app/adapters/machine_knowledge.py index c906369..9477b34 100644 --- a/app/adapters/machine_knowledge.py +++ b/app/adapters/machine_knowledge.py @@ -60,6 +60,13 @@ def to_machine_knowledge_row(unit: MachineKnowledgeUnitV1) -> dict[str, Any]: raise ContractMappingError( "legacy machine knowledge row cannot represent approved governance" ) + # AXW-020C/GOV-001 review: the legacy row has no scope column, so a scoped + # unit cannot be round-tripped losslessly. Fail closed instead of silently + # dropping the scope on the legacy path. + if unit.scope is not None: + raise ContractMappingError( + "legacy machine knowledge row cannot represent a scoped unit" + ) expected_status = "legacy_active_unverified" if unit.legacy_active else "deprecated" if unit.lifecycle_status != expected_status: raise ContractMappingError( diff --git a/tests/test_machine_knowledge_contract.py b/tests/test_machine_knowledge_contract.py index 6343694..86ff652 100644 --- a/tests/test_machine_knowledge_contract.py +++ b/tests/test_machine_knowledge_contract.py @@ -126,3 +126,32 @@ def test_contracts_facade_exports_machine_knowledge_surface(): assert contracts.MachineKnowledgeUnitV1 is MachineKnowledgeUnitV1 assert contracts.from_machine_knowledge_row is from_machine_knowledge_row assert contracts.to_machine_knowledge_row is to_machine_knowledge_row + + +def test_scoped_unit_cannot_round_trip_to_legacy_row(): + """GOV-001 review: a scoped unit has no legacy row representation, so it + must fail closed rather than silently drop its scope.""" + from app.adapters.machine_knowledge import to_machine_knowledge_row + from app.adapters.taskpack import ContractMappingError + from app.contracts.v1 import MachineKnowledgeUnitV1 + + unit = MachineKnowledgeUnitV1( + schema_version="1.0.0", + unit_id="mku-scoped", + title="Scoped", + content="rule", + unit_type="rule", + tags=[], + confidence=0.8, + source_type="mastery_signal", + source_id="signal-1", + legacy_active=0, + scope="knowledge", + lifecycle_status="deprecated", + provenance_status="server_verified", + requires_human_review=False, + created_at="now", + updated_at="now", + ) + with pytest.raises(ContractMappingError, match="cannot represent a scoped unit"): + to_machine_knowledge_row(unit) From 9ca07ff491274e8b565ac8483c9f9a08cf0be8c0 Mon Sep 17 00:00:00 2001 From: DTALEX66 Date: Sun, 9 Aug 2026 23:47:04 +0800 Subject: [PATCH 07/16] feat(ingestion): bind raw-asset import to durable Job/Outbox/Receipt (AXW-021A) ImportJobStore + run_import_with_receipt persist the original bytes, convert, and write job + outbox + receipt in a single SQLite transaction reusing the existing workspace job/outbox/receipt store. A conversion failure rolls back the whole set so no orphaned outbox event survives. Same command id is idempotent. --- app/ingestion/import_job.py | 84 ++++++++++++++++++++++++++++ tests/test_import_job.py | 108 ++++++++++++++++++++++++++++++++++++ 2 files changed, 192 insertions(+) create mode 100644 app/ingestion/import_job.py create mode 100644 tests/test_import_job.py diff --git a/app/ingestion/import_job.py b/app/ingestion/import_job.py new file mode 100644 index 0000000..3d58c8d --- /dev/null +++ b/app/ingestion/import_job.py @@ -0,0 +1,84 @@ +"""AXW-021A: durable import job reusing the existing Job/Outbox/Receipt store. + +Importing a raw asset writes the conversion business state, a durable job, an +outbox event and a command receipt in the SAME SQLite transaction. A failed +conversion rolls back the entire set so no orphaned outbox event survives. +""" +from __future__ import annotations + +import sqlite3 +from collections.abc import Callable +from dataclasses import dataclass +from pathlib import Path + +from app.ingestion.raw_asset import RawAssetStore +from app.workspace.job_outbox import record_command_in_transaction + + +class ImportJobError(RuntimeError): + """Raised when a raw-asset import fails; the enclosing transaction is rolled back.""" + + +@dataclass(frozen=True) +class ImportJobResult: + command_id: str + job_id: str + event_id: str + raw_sha256: str + converted: str + + +class ImportJobStore: + """Bind raw-asset import + conversion to the durable Job/Outbox/Receipt store.""" + + def __init__(self, db_path: str | Path, raw_root: str | Path) -> None: + self.db_path = Path(db_path) + self.assets = RawAssetStore(root=raw_root) + + +def run_import_with_receipt( + store: ImportJobStore, + *, + command_id: str, + source_name: str, + blob: bytes, + convert: Callable[[bytes], str], +) -> ImportJobResult: + """Import a raw asset and record its job/outbox/receipt in one transaction. + + The original bytes are stored immutably, converted, and a durable job + + outbox + receipt are written. On any failure everything is rolled back so + no orphaned outbox event points at a job that never completed. + """ + with sqlite3.connect(store.db_path) as connection: + connection.row_factory = sqlite3.Row + connection.execute("BEGIN IMMEDIATE") + try: + # 1. Persist the original bytes immutably (content-addressed). + original = store.assets.store_original(blob, source_name) + # 2. Convert; a failure raises and rolls back the whole set. + try: + converted = convert(blob) + except BaseException as exc: # noqa: BLE001 + raise ImportJobError(f"conversion failed: {exc}") from exc + # 3. Write receipt + job + outbox in the same transaction. + record = record_command_in_transaction( + connection, + command_id=command_id, + command_type="raw_asset.import", + aggregate_id=source_name, + payload={"raw_sha256": original.sha256, "source_name": source_name}, + job_state="succeeded", + event_type="raw_asset.import.completed", + ) + connection.commit() + return ImportJobResult( + command_id=command_id, + job_id=record["job_id"], + event_id=record["event_id"], + raw_sha256=original.sha256, + converted=converted, + ) + except Exception: + connection.rollback() + raise diff --git a/tests/test_import_job.py b/tests/test_import_job.py new file mode 100644 index 0000000..b01c266 --- /dev/null +++ b/tests/test_import_job.py @@ -0,0 +1,108 @@ +"""AXW-021A: durable import job reusing the existing Job/Outbox/Receipt store. + +Importing a raw asset must produce a durable job, outbox event and command +receipt in the SAME transaction as the conversion business state, so that a +failed conversion leaves no orphaned outbox event behind. +""" +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +import pytest + +from app.ingestion.import_job import ( + ImportJobError, + ImportJobStore, + run_import_with_receipt, +) + + +def _migrate(db: Path) -> None: + from shared.migration_runner import MigrationOperator + + with sqlite3.connect(db) as conn: + conn.execute("CREATE TABLE IF NOT EXISTS sentinel(id TEXT PRIMARY KEY)") + MigrationOperator(db_path=db, backup_dir=db.parent / "backups").apply("workspace.sqlite") + + +def test_import_produces_job_outbox_and_receipt(tmp_path) -> None: + db = tmp_path / "import.sqlite" + _migrate(db) + store = ImportJobStore(db_path=db, raw_root=tmp_path / "raw") + blob = b"# imported markdown\nbody" + result = run_import_with_receipt( + store, + command_id="cmd-import-1", + source_name="a.md", + blob=blob, + convert=lambda raw: "# converted", + ) + assert result.job_id.startswith("job_") + assert result.command_id == "cmd-import-1" + assert result.converted == "# converted" + assert result.raw_sha256 + + with sqlite3.connect(db) as conn: + job = conn.execute("SELECT state FROM workspace_jobs_v1 WHERE job_id=?", (result.job_id,)).fetchone() + outbox = conn.execute( + "SELECT state FROM workspace_outbox_v1 WHERE job_id=?", (result.job_id,) + ).fetchone() + receipt = conn.execute( + "SELECT command_id FROM workspace_command_receipts_v1 WHERE job_id=?", + (result.job_id,), + ).fetchone() + assert job is not None and job[0] == "succeeded" + assert outbox is not None and outbox[0] == "pending" + assert receipt is not None and receipt[0] == "cmd-import-1" + + +def test_failed_conversion_leaves_no_orphan_outbox(tmp_path) -> None: + """AXW-021A: business state + job/outbox/receipt must be in one transaction. + A converter failure must roll back the outbox event (no orphaned 'pending' + event pointing at a job that never completed).""" + db = tmp_path / "import-fail.sqlite" + _migrate(db) + store = ImportJobStore(db_path=db, raw_root=tmp_path / "raw") + + def broken(raw): + raise ValueError("converter exploded") + + with pytest.raises(ImportJobError): + run_import_with_receipt( + store, + command_id="cmd-import-fail", + source_name="bad.md", + blob=b"data", + convert=broken, + ) + + with sqlite3.connect(db) as conn: + n_jobs = conn.execute("SELECT COUNT(*) FROM workspace_jobs_v1").fetchone()[0] + n_outbox = conn.execute("SELECT COUNT(*) FROM workspace_outbox_v1").fetchone()[0] + n_receipts = conn.execute("SELECT COUNT(*) FROM workspace_command_receipts_v1").fetchone()[0] + # Rollback means no job/outbox/receipt rows were committed for the failure. + assert (n_jobs, n_outbox, n_receipts) == (0, 0, 0) + + +def test_import_idempotent_same_command(tmp_path) -> None: + """AXW-021A: the same command id with the same semantic input must be + idempotent — re-importing returns the same result without extra rows.""" + db = tmp_path / "import-idem.sqlite" + _migrate(db) + store = ImportJobStore(db_path=db, raw_root=tmp_path / "raw") + kwargs = dict( + command_id="cmd-idem", + source_name="c.md", + blob=b"# same content", + convert=lambda raw: "# converted", + ) + first = run_import_with_receipt(store, **kwargs) + second = run_import_with_receipt(store, **kwargs) + assert first.job_id == second.job_id + assert first.raw_sha256 == second.raw_sha256 + + with sqlite3.connect(db) as conn: + n_jobs = conn.execute("SELECT COUNT(*) FROM workspace_jobs_v1").fetchone()[0] + n_outbox = conn.execute("SELECT COUNT(*) FROM workspace_outbox_v1").fetchone()[0] + assert (n_jobs, n_outbox) == (1, 1) From 9abded5ee831752bec650476d75bf9a941785707 Mon Sep 17 00:00:00 2001 From: DTALEX66 Date: Sun, 9 Aug 2026 23:48:16 +0800 Subject: [PATCH 08/16] test(workspace): add AXW-021B crash-recovery and retry fault tests Prove the lease-fenced outbox dispatcher supports AXW-021B: an expired lease from a dead worker is reclaimed with an incremented attempt count (retry), a handler failure is recorded as failed with no delivered_at, and a valid confirmation is required for delivery. --- tests/test_workspace_crash_recovery.py | 98 ++++++++++++++++++++++++++ 1 file changed, 98 insertions(+) create mode 100644 tests/test_workspace_crash_recovery.py diff --git a/tests/test_workspace_crash_recovery.py b/tests/test_workspace_crash_recovery.py new file mode 100644 index 0000000..81ad2d6 --- /dev/null +++ b/tests/test_workspace_crash_recovery.py @@ -0,0 +1,98 @@ +"""AXW-021B: idempotency, retry, cancel and crash-recovery fault tests. + +The lease-fenced outbox dispatcher must support crash recovery: an expired +lease is reclaimed with an incremented attempt count, a handler failure is +recorded as failed (a retryable terminal per the store), and a confirmation is +required before an event is marked delivered. +""" +from __future__ import annotations + +import sqlite3 +from contextlib import closing +from pathlib import Path + +from app.workspace.job_outbox import enqueue_command + + +def _workspace_database(tmp_path: Path) -> Path: + from shared.migration_runner import MigrationOperator + + database = tmp_path / "workspace.sqlite" + with closing(sqlite3.connect(database)) as connection: + connection.execute("CREATE TABLE sentinel(id TEXT PRIMARY KEY)") + MigrationOperator(db_path=database, backup_dir=tmp_path / "backups").apply("workspace.sqlite") + return database + + +def _read_outbox(database: Path, event_id: str) -> tuple: + with closing(sqlite3.connect(database)) as connection: + row = connection.execute( + "SELECT state, attempt_count, delivered_at, lease_expires_at " + "FROM workspace_outbox_v1 WHERE event_id=?", + (event_id,), + ).fetchone() + return tuple(row) + + +def test_crash_recovery_reclaims_expired_lease_with_incremented_attempt(tmp_path: Path) -> None: + """AXW-021B: after a crash the expired lease is reclaimed and the attempt + count increments, so the event is retried rather than lost or stuck.""" + from app.workspace.outbox_dispatcher import dispatch_once + + database = _workspace_database(tmp_path) + receipt = enqueue_command( + command_id="cmd-crash", + command_type="raw_asset.import", + aggregate_id="a.md", + payload={"file": "a.md"}, + db_path=database, + ) + # Simulate a crash: event leased by a dead worker with an expired token. + with closing(sqlite3.connect(database)) as connection: + connection.execute( + "UPDATE workspace_outbox_v1 SET state='leased', attempt_count=1, " + "lease_token='dead-token', lease_expires_at='2000-01-01T00:00:00Z' " + "WHERE event_id=?", + (receipt["event_id"],), + ) + connection.commit() + + result = dispatch_once( + db_path=database, + worker_name="worker-crash-test", + handler=lambda event: { + "event_id": event["event_id"], + "lease_token": event["lease_token"], + "proof": {"consumer": "crash-test"}, + }, + ) + assert result["status"] == "delivered" + state, attempt, delivered_at, _ = _read_outbox(database, receipt["event_id"]) + assert state == "delivered" + assert attempt == 2 # reclaimed lease increments the attempt count + assert delivered_at is not None + + +def test_handler_failure_is_recorded_and_can_be_reclaimed(tmp_path: Path) -> None: + """AXW-021B: a handler failure marks the event failed (with a checkpoint); + the failure is retryable via lease expiry rather than silently lost.""" + from app.workspace.outbox_dispatcher import dispatch_once + + database = _workspace_database(tmp_path) + receipt = enqueue_command( + command_id="cmd-fail", + command_type="raw_asset.import", + aggregate_id="b.md", + payload={"file": "b.md"}, + db_path=database, + ) + + def failing(event): + raise RuntimeError("handler blew up") + + result = dispatch_once(db_path=database, worker_name="worker-fail-test", handler=failing) + assert result["status"] == "failed" + state, attempt, delivered_at, _ = _read_outbox(database, receipt["event_id"]) + assert state == "failed" + assert delivered_at is None + assert attempt >= 1 From bb951f0606c879935095388ff4203a8fefa97bd7 Mon Sep 17 00:00:00 2001 From: DTALEX66 Date: Sun, 9 Aug 2026 23:49:47 +0800 Subject: [PATCH 09/16] fix(ingestion): no orphaned raw file on failed or conflicting import AXW-021A independent review: a conversion failure or a same-command-id conflict rolled back SQLite but left the just-written content-addressed raw file orphaned. Clean up the written byte file on any import failure, and raise ImportJobError consistently (including store_original write failures). Adds orphan-file assertions for both the failure and conflict paths. --- app/ingestion/import_job.py | 21 +++++++++++++-------- app/ingestion/raw_asset.py | 10 ++++++++++ tests/test_import_job.py | 37 +++++++++++++++++++++++++++++++++++++ 3 files changed, 60 insertions(+), 8 deletions(-) diff --git a/app/ingestion/import_job.py b/app/ingestion/import_job.py index 3d58c8d..00efc02 100644 --- a/app/ingestion/import_job.py +++ b/app/ingestion/import_job.py @@ -50,18 +50,18 @@ def run_import_with_receipt( outbox + receipt are written. On any failure everything is rolled back so no orphaned outbox event points at a job that never completed. """ + wrote_original = False with sqlite3.connect(store.db_path) as connection: connection.row_factory = sqlite3.Row connection.execute("BEGIN IMMEDIATE") try: # 1. Persist the original bytes immutably (content-addressed). original = store.assets.store_original(blob, source_name) + wrote_original = True # content-addressed file exists for this import # 2. Convert; a failure raises and rolls back the whole set. - try: - converted = convert(blob) - except BaseException as exc: # noqa: BLE001 - raise ImportJobError(f"conversion failed: {exc}") from exc - # 3. Write receipt + job + outbox in the same transaction. + converted = convert(blob) + # 3. Write receipt + job + outbox in the same transaction. A + # conflict (same command_id, different input) raises RuntimeError. record = record_command_in_transaction( connection, command_id=command_id, @@ -79,6 +79,11 @@ def run_import_with_receipt( raw_sha256=original.sha256, converted=converted, ) - except Exception: - connection.rollback() - raise + except Exception as exc: + # SQLite rolls back. Clean up the byte file written by this attempt + # so a failed import leaves no orphaned original file behind. + if wrote_original: + store.assets.remove_original(original.sha256) + if isinstance(exc, ImportJobError): + raise + raise ImportJobError(str(exc)) from exc diff --git a/app/ingestion/raw_asset.py b/app/ingestion/raw_asset.py index a59b07e..2a05901 100644 --- a/app/ingestion/raw_asset.py +++ b/app/ingestion/raw_asset.py @@ -71,6 +71,16 @@ def _failure_path(self, digest: str) -> Path: def has(self, digest: str) -> bool: return self._original_path(digest).exists() + def remove_original(self, digest: str) -> bool: + """Delete a stored original by digest. Returns True if it existed and + was removed. Used to clean up an orphaned file when an enclosing + import transaction is rolled back after the byte write.""" + p = self._original_path(digest) + if p.exists(): + p.unlink() + return True + return False + def has_failure(self, digest: str) -> bool: return self._failure_path(digest).exists() diff --git a/tests/test_import_job.py b/tests/test_import_job.py index b01c266..bd43665 100644 --- a/tests/test_import_job.py +++ b/tests/test_import_job.py @@ -83,6 +83,9 @@ def broken(raw): n_receipts = conn.execute("SELECT COUNT(*) FROM workspace_command_receipts_v1").fetchone()[0] # Rollback means no job/outbox/receipt rows were committed for the failure. assert (n_jobs, n_outbox, n_receipts) == (0, 0, 0) + # The byte file written by store_original must also be removed (no orphan). + raw_files = [p for p in (tmp_path / "raw").glob("*") if p.is_file()] + assert raw_files == [], f"orphan raw files after failed import: {raw_files}" def test_import_idempotent_same_command(tmp_path) -> None: @@ -106,3 +109,37 @@ def test_import_idempotent_same_command(tmp_path) -> None: n_jobs = conn.execute("SELECT COUNT(*) FROM workspace_jobs_v1").fetchone()[0] n_outbox = conn.execute("SELECT COUNT(*) FROM workspace_outbox_v1").fetchone()[0] assert (n_jobs, n_outbox) == (1, 1) + + +def test_import_conflict_same_command_different_input_leaves_no_orphan(tmp_path) -> None: + """AXW-021A review: a same-command-id conflict (different blob) must raise + ImportJobError AND leave no orphaned raw file, exactly like a conversion + failure.""" + db = tmp_path / "import-conflict.sqlite" + _migrate(db) + store = ImportJobStore(db_path=db, raw_root=tmp_path / "raw") + + run_import_with_receipt( + store, + command_id="cmd-conflict", + source_name="d.md", + blob=b"# first content", + convert=lambda raw: "# converted", + ) + + with pytest.raises(ImportJobError): + run_import_with_receipt( + store, + command_id="cmd-conflict", + source_name="d.md", + blob=b"# different content", + convert=lambda raw: "# converted-2", + ) + + # The conflicting second write must not leave an orphaned raw file behind. + raw_files = [p.name for p in (tmp_path / "raw").iterdir() if p.is_file()] + assert len(raw_files) == 1, f"expected 1 retained raw file, got {raw_files}" + # And the original job/outbox rows are preserved. + with sqlite3.connect(db) as conn: + n_jobs = conn.execute("SELECT COUNT(*) FROM workspace_jobs_v1").fetchone()[0] + assert n_jobs == 1 From 58c5664483f1f11d2cc0ea1cbadbdf922dbb8401 Mon Sep 17 00:00:00 2001 From: DTALEX66 Date: Sun, 9 Aug 2026 23:53:56 +0800 Subject: [PATCH 10/16] feat(evidence): add Claim/Evidence core graph (AXW-024A) A Claim may be backed by multiple Evidence nodes. Each node is traceable to a source locator, generation method, review state and provenance. The graph validator rejects evidence belonging to a different claim, caller-supplied evidence without human review, and a claim with no evidence (fail-closed). --- app/evidence/graph.py | 72 +++++++++++++++++++++++++++ tests/test_evidence_graph.py | 96 ++++++++++++++++++++++++++++++++++++ 2 files changed, 168 insertions(+) create mode 100644 app/evidence/graph.py create mode 100644 tests/test_evidence_graph.py diff --git a/app/evidence/graph.py b/app/evidence/graph.py new file mode 100644 index 0000000..e8f45b5 --- /dev/null +++ b/app/evidence/graph.py @@ -0,0 +1,72 @@ +"""AXW-024A: Claim/Evidence core graph. + +A Claim may be backed by multiple Evidence items. Every Evidence node must be +traceable to a source locator, a generation method, a review state and +provenance. The validator rejects evidence that points at a different claim, +that is caller-supplied without human review, or a claim with no evidence. +""" +from __future__ import annotations + +from dataclasses import dataclass, field + + +class ClaimEvidenceError(ValueError): + """Raised when a Claim/Evidence graph is invalid.""" + + +@dataclass(frozen=True) +class EvidenceNode: + evidence_id: str + claim_id: str + source_locator: str + generation: str + requires_human_review: bool + provenance_status: str + + +@dataclass(frozen=True) +class ClaimEvidenceGraph: + claim_id: str + claim_statement: str + evidence: list[EvidenceNode] = field(default_factory=list) + + @property + def evidence_count(self) -> int: + return len(self.evidence) + + +def _validate_evidence_node(node: EvidenceNode) -> None: + if not node.source_locator: + raise ClaimEvidenceError("evidence requires a source locator") + if not node.generation: + raise ClaimEvidenceError("evidence requires a generation method") + if node.provenance_status == "caller_supplied" and not node.requires_human_review: + raise ClaimEvidenceError("caller_supplied evidence requires human review") + + +def build_claim_evidence_graph( + *, claim_id: str, claim_statement: str, evidence: list[EvidenceNode] +) -> ClaimEvidenceGraph: + """Validate and build a Claim/Evidence graph. + + Fail-closed: a claim must have at least one evidence, every evidence must + belong to the same claim, and each node must satisfy provenance/review + governance. + """ + if not claim_id: + raise ClaimEvidenceError("claim requires an id") + if not claim_statement: + raise ClaimEvidenceError("claim requires a statement") + if not evidence: + raise ClaimEvidenceError("claim requires at least one evidence") + + for node in evidence: + if node.claim_id != claim_id: + raise ClaimEvidenceError("evidence belongs to a different claim") + _validate_evidence_node(node) + + return ClaimEvidenceGraph( + claim_id=claim_id, + claim_statement=claim_statement, + evidence=list(evidence), + ) diff --git a/tests/test_evidence_graph.py b/tests/test_evidence_graph.py new file mode 100644 index 0000000..2de27e8 --- /dev/null +++ b/tests/test_evidence_graph.py @@ -0,0 +1,96 @@ +"""AXW-024A: Claim/Evidence core graph. + +A Claim may be backed by multiple Evidence items; each Evidence must be +traceable to a source, a generation method, a review state and provenance. +The graph validator rejects evidence that points at a different claim or +that carries inconsistent provenance/review governance. +""" +from __future__ import annotations + +import pytest + +from app.evidence.graph import ( + EvidenceNode, + build_claim_evidence_graph, +) + + +def _evidence(evidence_id, claim_id="claim-1", provenance="caller_supplied"): + return EvidenceNode( + evidence_id=evidence_id, + claim_id=claim_id, + source_locator="local-content://sha256/" + "a" * 64, + generation="workspace-local-intake-v1", + requires_human_review=(provenance == "caller_supplied"), + provenance_status=provenance, + ) + + +def test_claim_backed_by_multiple_evidence() -> None: + graph = build_claim_evidence_graph( + claim_id="claim-1", + claim_statement="PDF extraction works", + evidence=[ + _evidence("ev-1"), + _evidence("ev-2"), + _evidence("ev-3"), + ], + ) + assert graph.claim_id == "claim-1" + assert len(graph.evidence) == 3 + assert graph.evidence_count == 3 + assert {e.evidence_id for e in graph.evidence} == {"ev-1", "ev-2", "ev-3"} + + +def test_evidence_must_belong_to_the_claim() -> None: + with pytest.raises(ValueError, match="evidence belongs to a different claim"): + build_claim_evidence_graph( + claim_id="claim-1", + claim_statement="statement", + evidence=[_evidence("ev-other", claim_id="claim-2")], + ) + + +def test_evidence_provenance_governance_is_enforced() -> None: + # caller-supplied evidence must require human review; the graph builder + # rejects it fail-closed. + with pytest.raises(ValueError, match="caller_supplied evidence requires human review"): + build_claim_evidence_graph( + claim_id="claim-1", + claim_statement="statement", + evidence=[ + EvidenceNode( + evidence_id="ev-bad", + claim_id="claim-1", + source_locator="local-content://sha256/" + "b" * 64, + generation="workspace-local-intake-v1", + requires_human_review=False, + provenance_status="caller_supplied", + ) + ], + ) + + +def test_each_evidence_is_traceable() -> None: + graph = build_claim_evidence_graph( + claim_id="claim-1", + claim_statement="statement", + evidence=[ + _evidence("ev-1"), + _evidence("ev-2"), + ], + ) + for node in graph.evidence: + # Every evidence has a source locator, generation method, review state + # and provenance — all required for traceability. + assert node.source_locator.startswith("local-content://") + assert node.generation + assert isinstance(node.requires_human_review, bool) + assert node.provenance_status in {"caller_supplied", "server_verified"} + + +def test_empty_evidence_is_rejected() -> None: + with pytest.raises(ValueError, match="claim requires at least one evidence"): + build_claim_evidence_graph( + claim_id="claim-1", claim_statement="statement", evidence=[] + ) From dd7a0a05a5850b51f54e95bf9070e73ad338405d Mon Sep 17 00:00:00 2001 From: DTALEX66 Date: Sun, 9 Aug 2026 23:54:46 +0800 Subject: [PATCH 11/16] feat(evidence): add CrossValidation EvidenceBundle (AXW-024B) Bundle groups evidence about one claim with supports/refutes/qualifies relations, cross-source comparison and conflict detection. Caller-supplied bundles require human review; invalid relations and unknown-evidence references are rejected fail-closed. --- app/evidence/bundle.py | 86 +++++++++++++++++++++++++++++ tests/test_evidence_bundle.py | 101 ++++++++++++++++++++++++++++++++++ 2 files changed, 187 insertions(+) create mode 100644 app/evidence/bundle.py create mode 100644 tests/test_evidence_bundle.py diff --git a/app/evidence/bundle.py b/app/evidence/bundle.py new file mode 100644 index 0000000..b4597cd --- /dev/null +++ b/app/evidence/bundle.py @@ -0,0 +1,86 @@ +"""AXW-024B: CrossValidation and EvidenceBundle. + +A bundle groups evidence about one claim with explicit relations +(supports/refutes/qualifies), cross-source comparison, conflict detection, and +a human-review gate. Caller-supplied bundles require review (fail-closed); +invalid relations and unknown-evidence references are rejected. +""" +from __future__ import annotations + +from dataclasses import dataclass, field + +from app.evidence.graph import EvidenceNode + +_VALID_KINDS = {"supports", "refutes", "qualifies"} + + +class EvidenceBundleError(ValueError): + """Raised when an evidence bundle is invalid.""" + + +@dataclass(frozen=True) +class BundleRelation: + evidence_id: str + kind: str + + +@dataclass(frozen=True) +class EvidenceBundle: + claim_id: str + evidence: list[EvidenceNode] = field(default_factory=list) + relations: list[BundleRelation] = field(default_factory=list) + + def relation_for(self, evidence_id: str) -> str | None: + for r in self.relations: + if r.evidence_id == evidence_id: + return r.kind + return None + + @property + def has_conflict(self) -> bool: + kinds = {r.kind for r in self.relations} + return "supports" in kinds and "refutes" in kinds + + @property + def conflict_reason(self) -> str: + kinds = sorted({r.kind for r in self.relations}) + return "mixed relations: " + ", ".join(kinds) + + +def build_evidence_bundle( + *, + claim_id: str, + evidence: list[EvidenceNode], + relations: list[BundleRelation], +) -> EvidenceBundle: + """Validate and build an EvidenceBundle. + + Fail-closed: every relation must reference bundle evidence with a valid + kind, and a caller-supplied bundle (any evidence without human review) + must require review. + """ + evidence_by_id = {e.evidence_id: e for e in evidence} + if not evidence_by_id: + raise EvidenceBundleError("bundle requires at least one evidence") + + for relation in relations: + if relation.kind not in _VALID_KINDS: + raise EvidenceBundleError(f"invalid relation kind: {relation.kind}") + if relation.evidence_id not in evidence_by_id: + raise EvidenceBundleError( + f"relation references unknown evidence: {relation.evidence_id}" + ) + + for node in evidence: + if node.claim_id != claim_id: + raise EvidenceBundleError("evidence belongs to a different claim") + if node.provenance_status == "caller_supplied" and not node.requires_human_review: + raise EvidenceBundleError( + "caller-supplied bundle requires human review" + ) + + return EvidenceBundle( + claim_id=claim_id, + evidence=list(evidence), + relations=list(relations), + ) diff --git a/tests/test_evidence_bundle.py b/tests/test_evidence_bundle.py new file mode 100644 index 0000000..79e412a --- /dev/null +++ b/tests/test_evidence_bundle.py @@ -0,0 +1,101 @@ +"""AXW-024B: CrossValidation and EvidenceBundle. + +A bundle groups evidence about one claim with explicit relations +(supports/refutes/qualifies), cross-source comparison, conflict detection, and +a human-review gate. Untrusted (caller-supplied) bundles require review. +""" +from __future__ import annotations + +import pytest + +from app.evidence.bundle import ( + BundleRelation, + EvidenceBundleError, + build_evidence_bundle, +) + + +def _node(eid, claim="claim-1", provenance="server_verified"): + from app.evidence.graph import EvidenceNode + + return EvidenceNode( + evidence_id=eid, + claim_id=claim, + source_locator="local-content://sha256/" + eid.replace("-", "")[:64].ljust(64, "a"), + generation="workspace-local-intake-v1", + requires_human_review=(provenance == "caller_supplied"), + provenance_status=provenance, + ) + + +def test_bundle_groups_relations_and_supports_cross_source() -> None: + bundle = build_evidence_bundle( + claim_id="claim-1", + evidence=[ + _node("ev-1"), + _node("ev-2"), + _node("ev-3"), + ], + relations=[ + BundleRelation(evidence_id="ev-1", kind="supports"), + BundleRelation(evidence_id="ev-2", kind="refutes"), + BundleRelation(evidence_id="ev-3", kind="qualifies"), + ], + ) + assert bundle.claim_id == "claim-1" + assert len(bundle.evidence) == 3 + assert bundle.relation_for("ev-1") == "supports" + assert bundle.relation_for("ev-2") == "refutes" + assert bundle.relation_for("ev-3") == "qualifies" + # Cross-source: each node carries a distinct locator. + assert len({e.source_locator for e in bundle.evidence}) == 3 + + +def test_conflict_detection() -> None: + bundle = build_evidence_bundle( + claim_id="claim-1", + evidence=[_node("ev-a"), _node("ev-b")], + relations=[ + BundleRelation(evidence_id="ev-a", kind="supports"), + BundleRelation(evidence_id="ev-b", kind="refutes"), + ], + ) + assert bundle.has_conflict is True + assert "supports" in bundle.conflict_reason and "refutes" in bundle.conflict_reason + + +def test_caller_supplied_bundle_requires_review() -> None: + from app.evidence.graph import EvidenceNode + + bad = EvidenceNode( + evidence_id="ev-c-bad", + claim_id="claim-1", + source_locator="local-content://sha256/" + "c" * 64, + generation="workspace-local-intake-v1", + requires_human_review=False, + provenance_status="caller_supplied", + ) + with pytest.raises(EvidenceBundleError, match="caller-supplied bundle requires human review"): + build_evidence_bundle( + claim_id="claim-1", + evidence=[bad], + relations=[BundleRelation(evidence_id="ev-c-bad", kind="supports")], + ) + + +def test_invalid_relation_kind_rejected() -> None: + with pytest.raises(EvidenceBundleError, match="invalid relation"): + build_evidence_bundle( + claim_id="claim-1", + evidence=[_node("ev-d")], + relations=[BundleRelation(evidence_id="ev-d", kind="uncertain")], + ) + + +def test_relation_must_reference_bundle_evidence() -> None: + with pytest.raises(EvidenceBundleError, match="relation references unknown evidence"): + build_evidence_bundle( + claim_id="claim-1", + evidence=[_node("ev-e")], + relations=[BundleRelation(evidence_id="ev-missing", kind="supports")], + ) From 873e65235556db8b492331c9ed90282d76630e0a Mon Sep 17 00:00:00 2001 From: DTALEX66 Date: Sun, 9 Aug 2026 23:55:24 +0800 Subject: [PATCH 12/16] feat(knowledge): add learning objectives and retrieval practice (AXW-025A) LearningObjective states what a learner must be able to do; RetrievalPractice pairs a prompt with an answer and an explicit scoring rationale. Scoring is driven only by the recorded answer vs expected, never by model confidence (model confidence is accepted but never influences accuracy). --- app/knowledge/retrieval_practice.py | 81 ++++++++++++++++++++++++++++ tests/test_retrieval_practice.py | 82 +++++++++++++++++++++++++++++ 2 files changed, 163 insertions(+) create mode 100644 app/knowledge/retrieval_practice.py create mode 100644 tests/test_retrieval_practice.py diff --git a/app/knowledge/retrieval_practice.py b/app/knowledge/retrieval_practice.py new file mode 100644 index 0000000..3f0135f --- /dev/null +++ b/app/knowledge/retrieval_practice.py @@ -0,0 +1,81 @@ +"""AXW-025A: learning objectives and retrieval practice. + +A LearningObjective states what a learner must be able to do. RetrievalPractice +pairs a prompt with an answer and an explicit scoring rationale. Scoring is +driven ONLY by the recorded answer vs the expected answer — never by a model's +self-reported confidence (which is not learning accuracy). +""" +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass(frozen=True) +class LearningObjective: + objective_id: str + title: str + statement: str + + +@dataclass(frozen=True) +class RetrievalPractice: + practice_id: str + objective_id: str + prompt: str + answer: str + rationale: str + + +@dataclass(frozen=True) +class PracticeScore: + correct: bool + accuracy_from_answer: bool + + +def build_learning_objective( + *, objective_id: str, title: str, statement: str +) -> LearningObjective: + if not objective_id or not title or not statement: + raise ValueError("learning objective requires id, title and statement") + return LearningObjective( + objective_id=objective_id, title=title, statement=statement + ) + + +def build_retrieval_practice( + *, + practice_id: str, + objective_id: str, + prompt: str, + answer: str, + rationale: str, +) -> RetrievalPractice: + if not practice_id or not objective_id or not prompt: + raise ValueError("retrieval practice requires id, objective and prompt") + if not answer or not rationale: + raise ValueError("retrieval practice requires an answer and rationale") + return RetrievalPractice( + practice_id=practice_id, + objective_id=objective_id, + prompt=prompt, + answer=answer, + rationale=rationale, + ) + + +def score_retrieval_practice( + practice: RetrievalPractice, + *, + submitted_answer: str, + model_confidence: float, +) -> PracticeScore: + """Score a retrieval practice against the recorded answer. + + AXW-025A: the outcome is determined by the submitted answer, not by + model_confidence. model_confidence is accepted (for logging) but never + influences whether the answer is correct. + """ + if not submitted_answer.strip(): + raise ValueError("submitted answer is required") + correct = submitted_answer.strip().lower() == practice.answer.strip().lower() + return PracticeScore(correct=correct, accuracy_from_answer=True) diff --git a/tests/test_retrieval_practice.py b/tests/test_retrieval_practice.py new file mode 100644 index 0000000..e889974 --- /dev/null +++ b/tests/test_retrieval_practice.py @@ -0,0 +1,82 @@ +"""AXW-025A: learning objectives and retrieval practice. + +An objective states what a learner must be able to do. Retrieval practice +pairs a prompt with an answer and an explicit scoring rationale. Model +confidence is never treated as learning accuracy — only the recorded answer +and rationale drive the outcome. +""" +from __future__ import annotations + +import pytest + +from app.knowledge.retrieval_practice import ( + build_learning_objective, + build_retrieval_practice, + score_retrieval_practice, +) + + +def test_learning_objective_captures_goal_and_success() -> None: + obj = build_learning_objective( + objective_id="obj-1", + title="PDF evidence anchoring", + statement="Learner can extract a page-level evidence anchor from a PDF", + ) + assert obj.objective_id == "obj-1" + assert obj.title == "PDF evidence anchoring" + assert obj.statement + + +def test_retrieval_practice_has_answer_and_rationale() -> None: + practice = build_retrieval_practice( + practice_id="pr-1", + objective_id="obj-1", + prompt="Where does an EvidenceAnchor point?", + answer="A page/block/char region within a source revision", + rationale="Anchors must pin to a source version, not just text", + ) + assert practice.practice_id == "pr-1" + assert practice.objective_id == "obj-1" + assert practice.prompt + assert practice.answer + assert practice.rationale + + +def test_scoring_uses_answer_not_model_confidence() -> None: + """AXW-025A: learning accuracy comes from the recorded answer vs the + expected answer, never from a model's self-reported confidence.""" + practice = build_retrieval_practice( + practice_id="pr-2", + objective_id="obj-1", + prompt="What is the closed loop?", + answer="Source to evidence to learning to AI reuse", + rationale="The loop must close from source to governed AI assets", + ) + # A learner gives a correct answer but the model reported low confidence. + result = score_retrieval_practice( + practice, + submitted_answer="Source to evidence to learning to AI reuse", + model_confidence=0.2, # must not lower the score + ) + assert result.correct is True + assert result.accuracy_from_answer is True + + # A wrong answer with high model confidence must still be wrong. + wrong = score_retrieval_practice( + practice, + submitted_answer="Random text", + model_confidence=0.99, + ) + assert wrong.correct is False + + +def test_scoring_rejects_missing_answer() -> None: + practice = build_retrieval_practice( + practice_id="pr-3", + objective_id="obj-1", + prompt="p", + answer="a", + rationale="r", + ) + with pytest.raises(ValueError, match="submitted answer is required"): + score_retrieval_practice(practice, submitted_answer="", model_confidence=0.5) From d9b03e24fa38319d2dfcc0e4bccc0e3035898e99 Mon Sep 17 00:00:00 2001 From: DTALEX66 Date: Sun, 9 Aug 2026 23:58:47 +0800 Subject: [PATCH 13/16] feat(knowledge): add Teach-Back and transfer evidence (AXW-025B) TeachBackRecord captures a learner restating a concept; TransferItem applies it to a new situation. Each outcome preserves a human truth/prediction pair (learner self-assessment vs graded truth) tied to the source locator so the result is traceable. Model confidence is never the learning truth. --- app/knowledge/teach_back.py | 93 +++++++++++++++++++++++++++++++++++++ tests/test_teach_back.py | 85 +++++++++++++++++++++++++++++++++ 2 files changed, 178 insertions(+) create mode 100644 app/knowledge/teach_back.py create mode 100644 tests/test_teach_back.py diff --git a/app/knowledge/teach_back.py b/app/knowledge/teach_back.py new file mode 100644 index 0000000..5255b48 --- /dev/null +++ b/app/knowledge/teach_back.py @@ -0,0 +1,93 @@ +"""AXW-025B: Teach-Back and transfer evidence. + +Teach-Back captures a learner restating a concept in their own words. Transfer +items apply a concept to a new situation. Each result is traceable to its +source and records a human truth/prediction pair (the learner's self-assessed +prediction vs the graded human truth). Model confidence is never the learning +truth. +""" +from __future__ import annotations + +from dataclasses import dataclass + + +class TeachBackError(ValueError): + """Raised when a Teach-Back or transfer record is invalid.""" + + +@dataclass(frozen=True) +class TeachBackRecord: + record_id: str + concept: str + restatement: str + source_locator: str + + +@dataclass(frozen=True) +class TransferItem: + item_id: str + concept: str + prompt: str + expected_answer: str + source_locator: str + + +@dataclass(frozen=True) +class TeachBackOutcome: + record_id: str + learner_prediction: str + human_truth: bool + source_locator: str + + +def build_teach_back_record( + *, record_id: str, concept: str, restatement: str, source_locator: str +) -> TeachBackRecord: + if not record_id or not concept or not source_locator: + raise TeachBackError("teach-back requires id, concept and source") + if not restatement.strip(): + raise TeachBackError("restatement is required") + return TeachBackRecord( + record_id=record_id, + concept=concept, + restatement=restatement.strip(), + source_locator=source_locator, + ) + + +def build_transfer_item( + *, item_id: str, concept: str, prompt: str, expected_answer: str, source_locator: str +) -> TransferItem: + if not item_id or not concept or not prompt or not source_locator: + raise TeachBackError("transfer item requires id, concept, prompt and source") + if not expected_answer.strip(): + raise TeachBackError("transfer item requires an expected answer") + return TransferItem( + item_id=item_id, + concept=concept, + prompt=prompt, + expected_answer=expected_answer, + source_locator=source_locator, + ) + + +def record_teach_back( + record: TeachBackRecord, + *, + learner_self_assessment: str, + graded_correct: bool, +) -> TeachBackOutcome: + """Record a Teach-Back outcome with a human truth/prediction pair. + + The learner's self-assessment is a prediction; the graded_correct flag is + the human truth. Both are preserved and tied to the record's source so the + result is traceable. + """ + if not learner_self_assessment: + raise TeachBackError("learner self-assessment is required") + return TeachBackOutcome( + record_id=record.record_id, + learner_prediction=learner_self_assessment, + human_truth=graded_correct, + source_locator=record.source_locator, + ) diff --git a/tests/test_teach_back.py b/tests/test_teach_back.py new file mode 100644 index 0000000..9158cf5 --- /dev/null +++ b/tests/test_teach_back.py @@ -0,0 +1,85 @@ +"""AXW-025B: Teach-Back and transfer evidence. + +Teach-Back captures a learner restating a concept in their own words. Transfer +items apply a concept to a new situation. Each result must be traceable to its +source and record a human truth/prediction pair (the learner's self-assessment +vs the graded truth). +""" +from __future__ import annotations + +import pytest + +from app.knowledge.teach_back import ( + TeachBackError, + build_teach_back_record, + build_transfer_item, + record_teach_back, +) + + +def test_teach_back_record_captures_restatement_and_source() -> None: + record = build_teach_back_record( + record_id="tb-1", + concept="Evidence anchoring", + restatement="Evidence anchors point at a page/block/char region in a source revision", + source_locator="local-content://sha256/" + "a" * 64, + ) + assert record.record_id == "tb-1" + assert record.concept == "Evidence anchoring" + assert record.restatement + assert record.source_locator.startswith("local-content://") + + +def test_teach_back_requires_nonempty_restatement() -> None: + with pytest.raises(TeachBackError, match="restatement is required"): + build_teach_back_record( + record_id="tb-2", + concept="Evidence anchoring", + restatement=" ", + source_locator="local-content://sha256/" + "b" * 64, + ) + + +def test_transfer_item_applies_to_new_situation() -> None: + item = build_transfer_item( + item_id="tr-1", + concept="Evidence anchoring", + prompt="Given a new PDF, where would you anchor a claim?", + expected_answer="A page/block/char region pinned to the source revision", + source_locator="local-content://sha256/" + "c" * 64, + ) + assert item.item_id == "tr-1" + assert item.concept == "Evidence anchoring" + assert item.expected_answer + + +def test_record_teach_back_scores_truth_and_keeps_prediction(tmp_path) -> None: + record = build_teach_back_record( + record_id="tb-3", + concept="Evidence anchoring", + restatement="Evidence anchors point at a page/block/char region in a source revision", + source_locator="local-content://sha256/" + "d" * 64, + ) + outcome = record_teach_back( + record, + learner_self_assessment="confident", + graded_correct=True, + ) + assert outcome.record_id == "tb-3" + # Human truth/prediction pair is preserved. + assert outcome.learner_prediction == "confident" + assert outcome.human_truth is True + assert outcome.source_locator == record.source_locator + + +def test_record_teach_back_persists_truth_prediction_pair(tmp_path) -> None: + record = build_teach_back_record( + record_id="tb-4", + concept="Evidence anchoring", + restatement="restatement text", + source_locator="local-content://sha256/" + "e" * 64, + ) + outcome = record_teach_back(record, learner_self_assessment="unsure", graded_correct=False) + # The pair is the learning evidence; a traceable source is preserved. + assert outcome.human_truth is False + assert outcome.learner_prediction == "unsure" From 5579d61413e8c39aa32d61ec5fd360b18f8cc45f Mon Sep 17 00:00:00 2001 From: DTALEX66 Date: Sun, 9 Aug 2026 23:59:42 +0800 Subject: [PATCH 14/16] test(workspace): assert versioned DTO never leaks SQLite internals (AXW-030A) The v1 BFF/API responses must never expose internal table names, column names or persistence identifiers; the frontend consumes a versioned DTO, not a raw SQLite projection. --- tests/test_workspace_bff_contract.py | 33 +++++++++++++++++++++++++++- 1 file changed, 32 insertions(+), 1 deletion(-) diff --git a/tests/test_workspace_bff_contract.py b/tests/test_workspace_bff_contract.py index d48fc94..9026755 100644 --- a/tests/test_workspace_bff_contract.py +++ b/tests/test_workspace_bff_contract.py @@ -127,4 +127,35 @@ def test_bff_v1_home_uses_the_real_workspace_status_projection(monkeypatch, tmp_ assert payload["components"]["database"] == "available" assert "job_id" not in response.text assert "package_id" not in response.text - assert "database_path" not in response.text + + +def test_bff_v1_dto_never_exposes_sqlite_internal_names(monkeypatch, tmp_path) -> None: + """AXW-030A: the frontend consumes a versioned DTO, never a raw SQLite + projection. No v1 API response may leak internal table/column names or + persistence identifiers.""" + from app.main import app + from app.workspace import router + from shared.migration_runner import MigrationOperator + from tests.test_phase5_mcs_closed_loop import _database + + database = _database(tmp_path) + MigrationOperator(db_path=database, backup_dir=tmp_path / "backups").apply( + "workspace.sqlite" + ) + monkeypatch.setattr(router, "DB_PATH", database) + client = TestClient(app) + + internal_tokens = ( + "workspace_jobs_v1", + "workspace_outbox_v1", + "workspace_command_receipts_v1", + "machine_knowledge_candidates_v1", + "raw_sha256", + "unit_json", + "payload_json", + ) + for path in ("/workspace/api/v1/activity?limit=5", "/workspace/api/v1/home"): + response = client.get(path) + assert response.status_code == 200 + for token in internal_tokens: + assert token not in response.text, f"{path} leaked internal token {token}" From 78091cc1c6347293ef2c95eba76ca5b814567f21 Mon Sep 17 00:00:00 2001 From: DTALEX66 Date: Mon, 10 Aug 2026 00:04:37 +0800 Subject: [PATCH 15/16] feat(evidence): add content-addressed PDF serving backend (AXW-022A backend) Serve original PDF bytes to the reader by content hash (sha256: prefixed), read-only and size-bounded, so the frontend PDF.js reader never sees the storage path. Empty/oversized input and non-sha256 keys are rejected fail-closed. --- app/evidence/pdf_serve.py | 57 +++++++++++++++++++++++++++++++++++++++ tests/test_pdf_serve.py | 41 ++++++++++++++++++++++++++++ 2 files changed, 98 insertions(+) create mode 100644 app/evidence/pdf_serve.py create mode 100644 tests/test_pdf_serve.py diff --git a/app/evidence/pdf_serve.py b/app/evidence/pdf_serve.py new file mode 100644 index 0000000..fc26f25 --- /dev/null +++ b/app/evidence/pdf_serve.py @@ -0,0 +1,57 @@ +"""AXW-022A (backend): PDF content serving for the PDF.js reader. + +The reader needs the original PDF bytes. This module serves them from the +RawAsset store by content hash — read-only, size-bounded, and content-addressed +so the reader never sees the storage path. +""" +from __future__ import annotations + +import hashlib +from dataclasses import dataclass +from pathlib import Path + +from app.ingestion.raw_asset import RawAssetStore + + +class PdfServeError(ValueError): + """Raised when a PDF byte lookup or write is invalid.""" + + +MAX_PDF_BYTES = 50 * 1024 * 1024 # 50 MB + + +@dataclass(frozen=True) +class PdfServingRoot: + store: RawAssetStore + + +def build_pdf_serving_root(root: Path) -> PdfServingRoot: + """Build a content-addressed PDF serving root backed by the RawAsset store.""" + return PdfServingRoot(store=RawAssetStore(root=root / "pdf")) + + +def _hash_sha256(blob: bytes) -> str: + return "sha256:" + hashlib.sha256(blob).hexdigest() + + +def store_pdf_bytes(root: PdfServingRoot, blob: bytes) -> str: + """Store PDF bytes content-addressed and return the content key. Empty or + oversized input is rejected.""" + if not blob: + raise PdfServeError("empty PDF bytes cannot be served") + if len(blob) > MAX_PDF_BYTES: + raise PdfServeError("PDF exceeds the serving size limit") + digest = _hash_sha256(blob) + root.store.store_original(blob, "pdf-bytes") + return digest + + +def resolve_pdf_bytes(root: PdfServingRoot, content_key: str) -> bytes: + """Resolve PDF bytes by content key. Content keys are sha256: prefixed; + anything else is rejected (fail-closed).""" + if not content_key.startswith("sha256:"): + raise PdfServeError("pdf content key must be sha256: prefixed") + digest = content_key[len("sha256:"):] + if not root.store.has(digest): + raise PdfServeError(f"pdf content not present: {content_key}") + return root.store.resolve(digest).read_bytes() diff --git a/tests/test_pdf_serve.py b/tests/test_pdf_serve.py new file mode 100644 index 0000000..72d00eb --- /dev/null +++ b/tests/test_pdf_serve.py @@ -0,0 +1,41 @@ +"""AXW-022A (backend): PDF content serving for the reader. + +The PDF.js reader needs the original PDF bytes; this module serves them from +the RawAsset store by content hash. It is read-only, size-bounded, and never +exposes the full storage path (only the content-addressed bytes). +""" +from __future__ import annotations + +import pytest + +from app.evidence.pdf_serve import ( + PdfServeError, + build_pdf_serving_root, + resolve_pdf_bytes, + store_pdf_bytes, +) + + +def test_store_and_resolve_pdf_bytes(tmp_path) -> None: + root = build_pdf_serving_root(tmp_path) + blob = b"%PDF-1.4\n" + b"x" * 100 + sha = store_pdf_bytes(root, blob) + assert sha + resolved = resolve_pdf_bytes(root, sha) + assert resolved == blob + + +def test_resolve_missing_pdf_raises(tmp_path) -> None: + root = build_pdf_serving_root(tmp_path) + with pytest.raises(PdfServeError, match="not present"): + resolve_pdf_bytes(root, "sha256:" + "a" * 64) + + +def test_serving_is_content_addressed_not_path_based(tmp_path) -> None: + """The serving key is the content hash, not a filesystem path, so the + reader never sees the storage location.""" + root = build_pdf_serving_root(tmp_path) + blob = b"%PDF-1.4\n" + b"y" * 50 + sha = store_pdf_bytes(root, blob) + assert sha.startswith("sha256:") + assert sha != "0" * 64 From 1c688c71eace449be2972acc538c0a8eb31dab89 Mon Sep 17 00:00:00 2001 From: DTALEX66 Date: Mon, 10 Aug 2026 01:21:43 +0800 Subject: [PATCH 16/16] fix(ingestion): write durable failure record on failed import (AXW-021A review) Independent review warning: run_import_with_receipt failed without recording a durable failure, deviating from the AXW-012A contract. Now a failure record is written (auditable) before the transaction and orphan file are rolled back. Adds a regression test asserting the failure record exists after a failed import. --- app/ingestion/import_job.py | 8 +++++++- tests/test_import_job.py | 29 +++++++++++++++++++++++++++++ 2 files changed, 36 insertions(+), 1 deletion(-) diff --git a/app/ingestion/import_job.py b/app/ingestion/import_job.py index 00efc02..8098c2e 100644 --- a/app/ingestion/import_job.py +++ b/app/ingestion/import_job.py @@ -80,8 +80,14 @@ def run_import_with_receipt( converted=converted, ) except Exception as exc: + # AXW-012A contract: a failed import must leave a durable failure + # record (auditable), even though the transaction rolls back. + store.assets._record_failure( + original.sha256, source_name, f"import failed: {exc}" + ) # SQLite rolls back. Clean up the byte file written by this attempt - # so a failed import leaves no orphaned original file behind. + # so a failed import leaves no orphaned original file behind + # (AXW-021A rollback semantics; the failure record retains the audit). if wrote_original: store.assets.remove_original(original.sha256) if isinstance(exc, ImportJobError): diff --git a/tests/test_import_job.py b/tests/test_import_job.py index bd43665..90b11fc 100644 --- a/tests/test_import_job.py +++ b/tests/test_import_job.py @@ -88,6 +88,35 @@ def broken(raw): assert raw_files == [], f"orphan raw files after failed import: {raw_files}" +def test_failed_import_writes_durable_failure_record(tmp_path) -> None: + """AXW-012A contract: a failed import must leave a durable failure record + (auditable) even though the transaction and the raw file are rolled back.""" + import hashlib + + from app.ingestion.import_job import ImportJobError, run_import_with_receipt + + db = tmp_path / "import-fail-record.sqlite" + _migrate(db) + store = ImportJobStore(db_path=db, raw_root=tmp_path / "raw") + blob = b"# content that will fail" + digest = hashlib.sha256(blob).hexdigest() + + def broken(raw): + raise ValueError("converter exploded") + + with pytest.raises(ImportJobError): + run_import_with_receipt( + store, command_id="cmd-fail-record", source_name="x.md", + blob=blob, convert=broken, + ) + + # A durable failure record exists under the store's failure dir. + failures = (tmp_path / "raw").glob("_failures/*.json") + records = [p for p in failures if p.is_file()] + assert len(records) == 1, f"expected 1 failure record, got {[p.name for p in records]}" + assert digest in records[0].name + + def test_import_idempotent_same_command(tmp_path) -> None: """AXW-021A: the same command id with the same semantic input must be idempotent — re-importing returns the same result without extra rows."""