feat: ingest_log parallel-write phase (#205)

src/aelfrice/classification.py

@@ -40,6 +40,7 @@
     BELIEF_PREFERENCE,
     BELIEF_REQUIREMENT,
     BELIEF_TYPES,
+    INGEST_SOURCE_FILESYSTEM,
     LOCK_NONE,
     ONBOARD_STATE_PENDING,
     ORIGIN_AGENT_INFERRED,
@@ -500,6 +501,14 @@ def accept_classifications(
             skipped_existing += 1
             continue
         alpha, beta = get_source_adjusted_prior(c.belief_type, source)
+        # v2.0 #205 parallel-write: log the host-classified text.
+        store.record_ingest(
+            source_kind=INGEST_SOURCE_FILESYSTEM,
+            source_path=source,
+            raw_text=text,
+            derived_belief_ids=[bid],
+            ts=timestamp,
+        )
         store.insert_belief(
             Belief(
                 id=bid,

src/aelfrice/cli.py

@@ -60,6 +60,7 @@
 )
 from aelfrice.models import (
     BELIEF_FACTUAL,
+    INGEST_SOURCE_CLI_REMEMBER,
     LOCK_NONE,
     LOCK_USER,
     ORIGIN_AGENT_INFERRED,
@@ -795,6 +796,13 @@ def _cmd_lock(args: argparse.Namespace, out: object) -> int:
         existing = store.get_belief(bid)
         now = _utc_now_iso()
         if existing is None:
+            # v2.0 #205 parallel-write.
+            store.record_ingest(
+                source_kind=INGEST_SOURCE_CLI_REMEMBER,
+                raw_text=args.statement,
+                derived_belief_ids=[bid],
+                ts=now,
+            )
             store.insert_belief(Belief(
                 id=bid,
                 content=args.statement,

src/aelfrice/ingest.py

@@ -29,6 +29,7 @@
     ANCHOR_TEXT_MAX_LEN,
     CORROBORATION_SOURCE_TRANSCRIPT_INGEST,
     EDGE_DERIVED_FROM,
+    INGEST_SOURCE_FILESYSTEM,
     LOCK_NONE,
     ORIGIN_AGENT_INFERRED,
     Belief,
@@ -127,6 +128,17 @@ def _ingest_turn_ids(
                 session_id=session_id,
             )
             continue
+        # v2.0 #205 parallel-write: log the classifier input before
+        # materializing the belief. belief_id is deterministic on
+        # (source, sentence) so derived_belief_ids is known up-front.
+        store.record_ingest(
+            source_kind=INGEST_SOURCE_FILESYSTEM,
+            source_path=source,
+            raw_text=sentence,
+            derived_belief_ids=[belief_id],
+            session_id=session_id,
+            ts=ts,
+        )
         belief = Belief(
             id=belief_id,
             content=sentence,

src/aelfrice/mcp_server.py

@@ -54,6 +54,7 @@
 from aelfrice.models import (
     BELIEF_FACTUAL,
     CORROBORATION_SOURCE_MCP_REMEMBER,
+    INGEST_SOURCE_MCP_REMEMBER,
     LOCK_NONE,
     LOCK_USER,
     ORIGIN_USER_STATED,
@@ -214,6 +215,13 @@ def tool_lock(store: MemoryStore, *, statement: str) -> dict[str, Any]:
     existing = store.get_belief(bid)
     now = _utc_now_iso()
     if existing is None:
+        # v2.0 #205 parallel-write: log the user-stated raw text.
+        store.record_ingest(
+            source_kind=INGEST_SOURCE_MCP_REMEMBER,
+            raw_text=statement,
+            derived_belief_ids=[bid],
+            ts=now,
+        )
         store.insert_belief(Belief(
             id=bid,
             content=statement,

src/aelfrice/models.py

@@ -90,6 +90,28 @@
     CORROBORATION_SOURCE_HOOK_INGEST,
 })
 
+# v2.0 #205 ingest_log source_kind enum. Wire-format strings; do not
+# rename without a migration. Spec: docs/design/write-log-as-truth.md.
+INGEST_SOURCE_FILESYSTEM: Final[str] = "filesystem"
+INGEST_SOURCE_GIT: Final[str] = "git"
+INGEST_SOURCE_PYTHON_AST: Final[str] = "python_ast"
+INGEST_SOURCE_MCP_REMEMBER: Final[str] = "mcp_remember"
+INGEST_SOURCE_CLI_REMEMBER: Final[str] = "cli_remember"
+INGEST_SOURCE_FEEDBACK_LOOP_SYNTHESIS: Final[str] = "feedback_loop_synthesis"
+# `legacy_unknown` is reserved for migration: pre-v2.0 beliefs get
+# synthesized log rows at their `created_at` timestamp.
+INGEST_SOURCE_LEGACY_UNKNOWN: Final[str] = "legacy_unknown"
+
+INGEST_SOURCE_KINDS: Final[frozenset[str]] = frozenset({
+    INGEST_SOURCE_FILESYSTEM,
+    INGEST_SOURCE_GIT,
+    INGEST_SOURCE_PYTHON_AST,
+    INGEST_SOURCE_MCP_REMEMBER,
+    INGEST_SOURCE_CLI_REMEMBER,
+    INGEST_SOURCE_FEEDBACK_LOOP_SYNTHESIS,
+    INGEST_SOURCE_LEGACY_UNKNOWN,
+})
+
 # --- Onboard-session states ---
 # A polymorphic onboard handshake passes through exactly two persisted
 # states: `pending` after `start_onboard_session` records the scanner

src/aelfrice/replay.py

@@ -0,0 +1,100 @@
+"""v2.0 #205 ingest_log validation harness.
+
+Two checks per the spec at docs/design/write-log-as-truth.md:
+
+1. **Reachability** (cheap): every belief in the canonical store has at
+   least one ingest_log row that references its id in
+   `derived_belief_ids`. This is the v2.0 contract guarantee — no
+   orphan beliefs. Runs by default in `aelf doctor`.
+
+2. **Full equality** (expensive, opt-in): re-run classifier over each
+   `ingest_log.raw_text` and compare to canonical `beliefs`. This is
+   the v2.x flip-readiness probe. Stubbed in v2.0 first slice;
+   surfaced via `aelf doctor --replay` once the derivation function
+   is factored out.
+
+Per memo D5(C). Per memo D3, beliefs whose only log rows have
+`source_kind=legacy_unknown` are excluded from full-equality checks
+(they have no `raw_text` that the current classifier can re-derive).
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+from aelfrice.models import INGEST_SOURCE_LEGACY_UNKNOWN
+from aelfrice.store import MemoryStore
+
+
+@dataclass(frozen=True)
+class ReachabilityReport:
+    """Result of the reachability check.
+
+    `total_beliefs`: count of canonical beliefs in the store.
+    `reachable`: count of beliefs with ≥1 log row pointing at them.
+    `orphan_belief_ids`: beliefs with zero log rows. v2.0 contract
+        requires this to be empty for stores that started life on
+        v2.0; pre-v2.0 stores legitimately have orphans until the
+        legacy_unknown migration runs (not shipped in this slice).
+    """
+    total_beliefs: int
+    reachable: int
+    orphan_belief_ids: list[str] = field(default_factory=list)
+
+    @property
+    def all_reachable(self) -> bool:
+        return self.total_beliefs == self.reachable
+
+
+def check_log_reachability(store: MemoryStore) -> ReachabilityReport:
+    """Hypothesis-check the reachability contract.
+
+    For every belief in `store`, query `iter_ingest_log_for_belief`.
+    Any belief with zero log rows is an orphan — a violation of the
+    spec's acceptance criterion #1.
+
+    Cost: O(n_beliefs × n_log) in the linear-scan implementation
+    (`iter_ingest_log_for_belief` walks all log rows). Acceptable for
+    a doctor-tier check; the validation harness is not on the
+    interactive path.
+    """
+    belief_ids = store.list_belief_ids()
+    orphans: list[str] = []
+    reachable = 0
+    for bid in belief_ids:
+        if store.iter_ingest_log_for_belief(bid):
+            reachable += 1
+        else:
+            orphans.append(bid)
+    return ReachabilityReport(
+        total_beliefs=len(belief_ids),
+        reachable=reachable,
+        orphan_belief_ids=orphans,
+    )
+
+
+@dataclass(frozen=True)
+class FullEqualityReport:
+    """Stub. v2.0 first slice does not implement full-equality replay.
+
+    Wired through so callers can detect "not implemented" without
+    raising; the spec's acceptance criterion #3 is partially met by
+    reachability, with full-equality landing in v2.x.
+    """
+    implemented: bool
+    excluded_legacy_unknown: int
+
+
+def replay_full_equality(store: MemoryStore) -> FullEqualityReport:
+    """v2.x flip-readiness probe. Not implemented in v2.0 first slice.
+
+    Returns `implemented=False` plus a count of legacy_unknown log
+    rows that would be excluded from the comparison anyway. The
+    intent is documented so a reviewer can grep this surface for the
+    next slice's wiring.
+    """
+    cur = store._conn.execute(  # pyright: ignore[reportPrivateUsage]
+        "SELECT COUNT(*) AS n FROM ingest_log WHERE source_kind = ?",
+        (INGEST_SOURCE_LEGACY_UNKNOWN,),
+    )
+    legacy_n = int(cur.fetchone()["n"])
+    return FullEqualityReport(implemented=False, excluded_legacy_unknown=legacy_n)

src/aelfrice/scanner.py

@@ -27,7 +27,12 @@
 
 from aelfrice.classification import classify_sentence
 from aelfrice.inedible import is_inedible
-from aelfrice.models import LOCK_NONE, ORIGIN_AGENT_INFERRED, Belief
+from aelfrice.models import (
+    INGEST_SOURCE_FILESYSTEM,
+    LOCK_NONE,
+    ORIGIN_AGENT_INFERRED,
+    Belief,
+)
 from aelfrice.noise_filter import NoiseConfig, is_noise
 from aelfrice.store import MemoryStore
 
@@ -233,6 +238,17 @@ def scan_repo(
             skipped_existing += 1
             continue
         created_at = candidate.commit_date or timestamp
+        # v2.0 #205 parallel-write: log the raw classifier input
+        # before materializing the belief. derived_belief_ids is
+        # known up-front because belief_id is deterministic on
+        # (source, text).
+        store.record_ingest(
+            source_kind=INGEST_SOURCE_FILESYSTEM,
+            source_path=candidate.source,
+            raw_text=candidate.text,
+            derived_belief_ids=[belief_id],
+            ts=created_at,
+        )
         store.insert_belief(Belief(
             id=belief_id,
             content=candidate.text,