diff --git a/CHANGELOG/v3.md b/CHANGELOG/v3.md index 8c5324bfb..a8c719301 100644 --- a/CHANGELOG/v3.md +++ b/CHANGELOG/v3.md @@ -12,6 +12,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **Dispatched subagents now inherit memory context ([#1068](https://github.com/robotrocketscience/aelfrice/issues/1068)).** Workers spawned via agent dispatch previously ran blind to the belief store — no SessionStart baseline, no per-prompt retrieval, and no locked constraints crossed the dispatch boundary. A new `PreToolUse:^(Agent|Task)$` hook (`aelf-agent-context-hook`) rewrites the worker's prompt through the harness's `updatedInput` channel, prepending a bounded `` block: L0 locked beliefs (with #1016 provenance-aware framing and reference-lock manifest bounding) plus beliefs relevant to the worker's task at the reduced auxiliary budget the Grep|Glob lane uses. Both harness channels were probed live before implementation (see the issue): `updatedInput` applies without a `permissionDecision` — so the hook never touches the user's permission flow — while `SubagentStart` was rejected for this lane (its payload carries no prompt text, so it cannot do query-aware retrieval). Fail-open everywhere (no store / malformed payload / retrieval error → byte-identical passthrough), idempotent on already-tagged prompts (nested dispatch never double-injects), and deterministic per #605. Installed by default via `aelf setup` (skip with `--no-agent-context`); runtime kill switch `AELFRICE_AGENT_CONTEXT=0`. Claude-host only for now — the Codex desired-set is unchanged pending #1056 live validation. +- **Temporal spine: ingest-time chronological edges + a dedicated retrieval lane, default-off ([#1064](https://github.com/robotrocketscience/aelfrice/issues/1064)).** The largest retrieval-coverage gain measured on this codebase, deterministic and embeddings-free: per-session `TEMPORAL_NEXT` chains written at ingest (`[ingest] write_temporal_spine` / `AELFRICE_TEMPORAL_SPINE_WRITE`), an idempotent `aelf spine backfill` for existing stores plus an `aelf doctor` spine row, and a retrieval lane (`[retrieval] use_temporal_spine` / `AELFRICE_TEMPORAL_SPINE`, node budget via `temporal_spine_budget`) that traverses the chains from the top-5 L1 seeds and appends chronological neighbours after L1. Reaches gold that shares zero salient terms with the question through chronological adjacency: confirmed **+14.6pp** gold-set coverage on LoCoMo (0.460 → 0.606; temporal +17.2pp, multi-hop +10.4pp, 10× the shuffled control), out-of-sample gain exceeding dev (+12.7pp on LongMemEval). Both flags land default-OFF; the default-ON flip is gated on the pre-registered criteria in `docs/design/feature-temporal-spine.md` (G2 production-budget trim survival, G3 latency, G4 migration, G5 determinism). `LaneTelemetry` gains `temporal_spine` + `temporal_spine_candidates`; `benchmarks/temporal_spine_ablation.py` is the lane's permanent coverage ablation (baseline / +spine / seeded shuffled-control). + - **Codex host target for `setup` / `doctor` / `unsetup` ([#1052](https://github.com/robotrocketscience/aelfrice/issues/1052)).** `aelf setup --host codex` writes the portable aelfrice hook subset into `~/.codex/hooks.json` — merge-aware, idempotent, and refusing (without `--force`) to clobber an unparseable existing file — and prints the trust-approval steps (`/hooks` in a Codex session; `codex features enable codex_hooks`, since the upstream hooks surface is feature-flagged off by default). `aelf doctor --host codex` validates hooks.json shape, aelfrice event coverage, on-disk commands, the feature flag, and `[hooks.state]` trust coverage; `aelf unsetup --host codex` removes only aelfrice-owned entries. Trust hashes are never computed client-side — the upstream trust schema is explicitly slated to change. Compaction rebuild rides `SessionStart(source=="compact")`, the only injection channel Codex honors ([#1054](https://github.com/robotrocketscience/aelfrice/issues/1054) investigation). Bash-matcher tool hooks (memory-first shell search, pre-issue duplicate guard, commit ingest) ride along unchanged ([#1055](https://github.com/robotrocketscience/aelfrice/issues/1055)): Codex canonicalizes shell commands to `tool_name == "Bash"`, so the matchers are host-portable; only the `Grep|Glob` hook (tools that don't exist on Codex) and the host-specific memory mirror are excluded. - **Persistent host-level auto-install opt-out ([#1053](https://github.com/robotrocketscience/aelfrice/issues/1053)).** A Codex-primary machine no longer needs `AELFRICE_NO_AUTO_INSTALL=1` inlined on every hook command: a new `opt_out_hosts` key in `~/.aelfrice/opt-out-hooks.json` (written automatically by `aelf setup --host codex` when the Claude host has no aelfrice hooks) gates the CLI-entry settings merge. Dual-host machines are left untouched; an explicit `aelf setup` clears the opt-out. diff --git a/benchmarks/temporal_spine_ablation.py b/benchmarks/temporal_spine_ablation.py new file mode 100644 index 000000000..ccc68ce0e --- /dev/null +++ b/benchmarks/temporal_spine_ablation.py @@ -0,0 +1,301 @@ +"""#1064 temporal-spine lane ablation — gold-set evidence coverage on LoCoMo. + +Runs LoCoMo under three retrieval configurations to isolate the +``use_temporal_spine`` lane: + + baseline — spine lane off (production-style default) + +spine — spine lane on, real chronological chains + shuffled-control — spine lane on, identical edge count but endpoints + deterministically permuted (chronology destroyed) + +Scoring is **gold-set coverage**: ``|gold ∩ retrieved| / |gold|`` per +question, where the gold set is the belief ids derived from the QA pair's +evidence dialogue turns. This is the lens the #1064 campaign pre-registered — +single-hit recall stays saturated and misses the aggregation/temporal +questions whose gold is diffuse. Also reported: the all-evidence rate +(fraction of questions whose gold set is fully covered). No LLM reader is +involved; the run is deterministic end-to-end. + +The shuffled control isolates the cause: identical density with scrambled +endpoints recovering ~nothing means the value is the chronology, not the +extra connectivity. The permutation is seeded (``--shuffle-seed``) so the +control is reproducible. + +Ingest is arm-independent (the flag only changes retrieval), so each +conversation is ingested once and the spine backfilled once; the shuffled +arm rewrites the TEMPORAL_NEXT edge set in place and restores it after. + +Usage: + uv run python -m benchmarks.temporal_spine_ablation \\ + --data /tmp/LoCoMo/data/locomo10.json \\ + --out /tmp/temporal_spine_ablation.json + # smoke: --subset-convs 1 --subset-qa 20 +""" +from __future__ import annotations + +import argparse +import json +import random +import time +from dataclasses import dataclass, field +from typing import Final + +from aelfrice.ingest import _ingest_turn_ids +from aelfrice.models import EDGE_TEMPORAL_NEXT +from aelfrice.retrieval import retrieve_v2 +from aelfrice.store import MemoryStore +from aelfrice.temporal_spine import backfill_temporal_spine + +from benchmarks.locomo_adapter import ( + CATEGORY_NAMES, + DEFAULT_DATA_PATH, + LoCoMoConversation, + _parse_locomo_datetime, + load_locomo, +) + +ARM_BASELINE: Final[str] = "baseline" +ARM_SPINE: Final[str] = "+spine" +ARM_SHUFFLED: Final[str] = "shuffled-control" +ARMS: Final[tuple[str, ...]] = (ARM_BASELINE, ARM_SPINE, ARM_SHUFFLED) + +# Wide-retrieval operating point of the #1064 campaign (dev ran +# l1_limit=200 / budget=8000; the confirmatory inherited it). The +# production operating point (1500-token hook budget) is the G2 +# flip-gate question, run via --budget / --l1-limit overrides. +DEFAULT_BUDGET: Final[int] = 8000 +DEFAULT_L1_LIMIT: Final[int] = 200 + + +@dataclass +class CoverageAccumulator: + """Per-arm coverage aggregates.""" + + n_questions: int = 0 + coverage_sum: float = 0.0 + n_all_evidence: int = 0 + per_category: dict[int, list[float]] = field(default_factory=dict) + + def add(self, category: int, coverage: float) -> None: + self.n_questions += 1 + self.coverage_sum += coverage + if coverage >= 1.0: + self.n_all_evidence += 1 + self.per_category.setdefault(category, []).append(coverage) + + def overall(self) -> float: + return ( + self.coverage_sum / self.n_questions if self.n_questions else 0.0 + ) + + def all_evidence_rate(self) -> float: + return ( + self.n_all_evidence / self.n_questions if self.n_questions else 0.0 + ) + + +def ingest_with_evidence_map( + store: MemoryStore, conv: LoCoMoConversation, +) -> dict[str, list[str]]: + """Ingest a conversation, returning ``{dia_id: [belief_ids]}``. + + Mirrors ``locomo_adapter.ingest_conversation`` (one store session per + LoCoMo session, date-prefixed turn text, parsed created_at) but records + which belief ids each dialogue turn derived — the mapping the coverage + scorer needs to turn evidence dia_ids into gold belief sets. Uses the + internal ``_ingest_turn_ids`` because the public ``ingest_turn`` + returns a count only. + """ + evidence_map: dict[str, list[str]] = {} + for session in conv.sessions: + am_session = store.create_session( + model="locomo-benchmark", + project_context=f"{conv.sample_id} session {session.session_num}", + ) + created = _parse_locomo_datetime(session.date_time) + if session.date_time: + _ingest_turn_ids( + store=store, + text=f"[Session {session.session_num}, {session.date_time}]", + source="locomo", + session_id=am_session.id, + created_at=created, + ) + for turn in session.turns: + ids = _ingest_turn_ids( + store=store, + text=f"[{session.date_time}] {turn.speaker}: {turn.text}", + source="locomo", + session_id=am_session.id, + created_at=created, + ) + if turn.dia_id: + evidence_map[turn.dia_id] = list(ids) + store.complete_session(am_session.id) + return evidence_map + + +def shuffle_spine_edges(store: MemoryStore, *, seed: int) -> int: + """Replace TEMPORAL_NEXT edges with an endpoint-permuted set. + + Same edge count, same node population, chronology destroyed: dst + endpoints are permuted across edges with a seeded Fisher-Yates, and + self-loops / duplicate triples are skipped (the tiny count lost to + collisions is reported by the return value so the arms stay honest). + Returns the number of edges written. + """ + conn = store._conn # noqa: SLF001 — bench-only surgical rewrite + rows = conn.execute( + "SELECT src, dst, weight FROM edges WHERE type = ?", + (EDGE_TEMPORAL_NEXT,), + ).fetchall() + srcs = [str(r["src"]) for r in rows] + dsts = [str(r["dst"]) for r in rows] + weights = [float(r["weight"]) for r in rows] + rng = random.Random(seed) + rng.shuffle(dsts) + conn.execute("DELETE FROM edges WHERE type = ?", (EDGE_TEMPORAL_NEXT,)) + written = 0 + seen: set[tuple[str, str]] = set() + for src, dst, weight in zip(srcs, dsts, weights): + if src == dst or (src, dst) in seen: + continue + seen.add((src, dst)) + conn.execute( + "INSERT INTO edges (src, dst, type, weight) VALUES (?, ?, ?, ?)", + (src, dst, EDGE_TEMPORAL_NEXT, weight), + ) + written += 1 + conn.commit() + return written + + +def restore_spine_edges(store: MemoryStore) -> int: + """Drop all TEMPORAL_NEXT edges and rebuild the real spine.""" + conn = store._conn # noqa: SLF001 + conn.execute("DELETE FROM edges WHERE type = ?", (EDGE_TEMPORAL_NEXT,)) + conn.commit() + report = backfill_temporal_spine(store) + return report.n_edges_written + + +def run_arm_on_store( + store: MemoryStore, + conv: LoCoMoConversation, + evidence_map: dict[str, list[str]], + arm: str, + acc: CoverageAccumulator, + *, + budget: int, + l1_limit: int, + subset_qa: int | None, +) -> None: + qa_pairs = conv.qa_pairs[:subset_qa] if subset_qa else conv.qa_pairs + lane_on = arm != ARM_BASELINE + for qa in qa_pairs: + gold: set[str] = { + bid + for dia_id in qa.evidence + for bid in evidence_map.get(dia_id, []) + } + if not gold: + continue # no scorable evidence (e.g. category-5 adversarial) + result = retrieve_v2( + store, + qa.question, + budget=budget, + l1_limit=l1_limit, + include_locked=False, + use_temporal_spine=lane_on, + ) + retrieved = {b.id for b in result.beliefs} + acc.add(qa.category, len(gold & retrieved) / len(gold)) + + +def main() -> None: + ap = argparse.ArgumentParser(description=__doc__) + ap.add_argument("--data", default=DEFAULT_DATA_PATH) + ap.add_argument("--out", default="/tmp/temporal_spine_ablation.json") + ap.add_argument("--budget", type=int, default=DEFAULT_BUDGET) + ap.add_argument("--l1-limit", type=int, default=DEFAULT_L1_LIMIT) + ap.add_argument("--shuffle-seed", type=int, default=1064) + ap.add_argument("--subset-convs", type=int, default=None) + ap.add_argument("--subset-qa", type=int, default=None) + args = ap.parse_args() + + conversations = load_locomo(args.data) + if args.subset_convs: + conversations = conversations[: args.subset_convs] + + accs: dict[str, CoverageAccumulator] = { + arm: CoverageAccumulator() for arm in ARMS + } + started = time.time() + spine_edges_total = 0 + for i, conv in enumerate(conversations): + store = MemoryStore(":memory:") + try: + evidence_map = ingest_with_evidence_map(store, conv) + spine_report = backfill_temporal_spine(store) + spine_edges_total += spine_report.n_edges_written + for arm in ARMS: + if arm == ARM_SHUFFLED: + shuffle_spine_edges(store, seed=args.shuffle_seed) + run_arm_on_store( + store, conv, evidence_map, arm, accs[arm], + budget=args.budget, l1_limit=args.l1_limit, + subset_qa=args.subset_qa, + ) + if arm == ARM_SHUFFLED: + restore_spine_edges(store) + finally: + store.close() + print( + f"[{i + 1}/{len(conversations)}] {conv.sample_id}: " + + ", ".join( + f"{arm}={accs[arm].overall():.3f}" for arm in ARMS + ) + ) + + report: dict[str, object] = { + "bench": "temporal_spine_ablation", + "issue": 1064, + "data": args.data, + "budget": args.budget, + "l1_limit": args.l1_limit, + "shuffle_seed": args.shuffle_seed, + "n_conversations": len(conversations), + "spine_edges_built": spine_edges_total, + "elapsed_seconds": round(time.time() - started, 1), + "arms": { + arm: { + "n_questions": acc.n_questions, + "coverage_overall": round(acc.overall(), 4), + "all_evidence_rate": round(acc.all_evidence_rate(), 4), + "coverage_by_category": { + CATEGORY_NAMES.get(cat, str(cat)): round( + sum(vals) / len(vals), 4, + ) + for cat, vals in sorted(acc.per_category.items()) + }, + } + for arm, acc in accs.items() + }, + } + with open(args.out, "w", encoding="utf-8") as f: + json.dump(report, f, indent=2) + + base = accs[ARM_BASELINE].overall() + spine = accs[ARM_SPINE].overall() + shuffled = accs[ARM_SHUFFLED].overall() + print(f"\nbaseline coverage : {base:.4f}") + print(f"+spine coverage : {spine:.4f} (Δ {spine - base:+.4f})") + print( + f"shuffled-control coverage : {shuffled:.4f} " + f"(Δ {shuffled - base:+.4f})" + ) + print(f"report -> {args.out}") + + +if __name__ == "__main__": + main() diff --git a/docs/design/feature-temporal-spine.md b/docs/design/feature-temporal-spine.md new file mode 100644 index 000000000..f89f817f0 --- /dev/null +++ b/docs/design/feature-temporal-spine.md @@ -0,0 +1,126 @@ +# Temporal spine — ingest-time chronological edges + dedicated retrieval lane (#1064) + +Status: **landed default-off** (writer + backfill + lane). The default-ON +flip is gated on the pre-registered criteria in § Flip gate below. + +## Mechanism + +Three components, all deterministic and embeddings-free (#605): + +1. **Spine writer** (`aelfrice.temporal_spine.write_temporal_spine`, + wired into ingest behind `[ingest] write_temporal_spine` / + `AELFRICE_TEMPORAL_SPINE_WRITE`, default off). After each belief + insert, link to the previous belief in the same `session_id` + (order `created_at`, tie-break insertion order) with + `TEMPORAL_NEXT`, src = successor, dst = predecessor (the + `models.py` semantics #386 settled), weight 0.8. One edge per + belief (~1.0 edges/belief measured), O(1) per insert. + +2. **Backfill** (`aelf spine backfill`, hidden subcommand). Idempotent + per-session chain build over an existing store (insert-if-absent per + `(src, dst, type)`), `--dry-run` counting mode, plus an `aelf + doctor` row (spine present/absent, edge count). Existing stores + predate the writer; the migration story cannot be "re-ingest + everything". Pinned by test: backfill output == writer output on the + same corpus. + +3. **Retrieval lane** (`use_temporal_spine` / + `AELFRICE_TEMPORAL_SPINE`, default off). Additive candidate source + after L1: traverses `TEMPORAL_NEXT` from the top-5 packed L1 seeds, + both directions, depth 1, node budget 32 + (`temporal_spine_budget` / `AELFRICE_TEMPORAL_SPINE_BUDGET`). + Appended after L1 candidates — never displaces them pre-packing. + No-op guard via `count_edges_by_type()`: zero spine edges → empty + lane at ~zero cost, byte-identical output for spineless stores. + Telemetry: `LaneTelemetry.temporal_spine` (packed survivors) + + `temporal_spine_candidates` (pre-pack discoveries); the delta is the + trim loss the G2 gate asks about. Soft-deleted beliefs + (`valid_to` set) are skip-but-continue at traversal time so chain + integrity survives GC; spine hits do **not** seed BFS (unmeasured + surface — the confirmatory evidence ran depth-1 append-after-L1 + with BFS untouched). + +## Why this works — and why it's a lane, not a gate change + +Metric: gold-set coverage = |gold ∩ retrieved| / |gold| per question +(the lens that matters for aggregation/temporal questions with diffuse +gold; single-hit recall stays saturated and misses this). Prior rounds +established that ~84% of the gold missing at wide config shares **zero** +salient terms with the question — unreachable by any lexical means. The +spine reaches that gold through chronological adjacency to beliefs that +*do* match. + +- **Dev (LongMemEval, 475 questions, l1=200/budget=8000):** overall + coverage 0.531 → 0.658 (+12.7pp); temporal-reasoning +13.6pp; + multi-session +11.1pp. A shuffled control — identical edge count, + endpoints permuted — gains ~nothing (311× ratio). Deterministic rerun + byte-identical. +- **Confirmatory (LoCoMo, 1,979 QA with evidence-belief gold sets, one + shot, criteria pre-registered):** overall **+14.6pp** (0.460 → 0.606); + temporal +17.2pp; multi-hop +10.4pp; 10× the shuffled control; + all-evidence rate 0.133 → 0.260. Out-of-sample gain exceeded dev. + Expansion-node budget curve is monotone (0.605 / 0.631 / 0.659 at + 32/64/128, ~+2.5pp per doubling, no plateau) — the effect is + budget-limited, not substrate-limited. + +The permanent ablation for this lane is +`benchmarks/temporal_spine_ablation.py` (gold-evidence coverage on +LoCoMo; arms: baseline / +spine / seeded shuffled-control). + +Why a lane and not a #741 gate exception: #977's keep-off verdict is +correct *for generic BFS* and stays untouched; #741 explicitly +out-scoped per-edge-type gating; and the default BFS knobs structurally +suppress temporal traversal (`BFS_EDGE_WEIGHTS[TEMPORAL_NEXT] = 0.25` × +`min_path_score = 0.10` prunes ≥2-hop chains, and temporal ranks last +per hop). The measured gains happened under depth-1-only traversal — +they are a floor. + +Distinct from the #998 A4 ratified decline: A4 declined token-Jaccard +*co-occurrence* edges fed through the #981 HRR-expand lane ("density is +a liability"). Different edge class (similarity vs chronology), +different consumer. The shuffled control is the direct answer to the +density concern: identical density with scrambled endpoints recovers ++1.5pp vs the spine's +14.6pp — the value is the chronology, not the +density. + +## Flip gate — pre-registered default-ON criteria + +Default-off is the **landing posture, not the end state**. When all +pass, the next release flips both flags (writer + lane) default-ON in +one release, with the backfill path included for existing stores: + +- **G1 — confirmatory evidence:** DONE (above; recorded in #1064). +- **G2 — production operating point:** coverage delta + top-rank + invariance at the production hook budget (1500 tokens) on bench pools + and a shadow eval on a real backfilled store (aggregate-only). Pass: + ≥ +3pp coverage at production budget, no top-rank regression. + (Dev/confirmatory ran at budget 8000; #1045/#1062 established the + trim binds in ≥21% of real injections — this gate answers "does it + survive the trim." Read `LaneTelemetry.temporal_spine_candidates − + temporal_spine` for the per-call trim loss.) +- **G3 — latency delta (#739-style):** with spine present at ≥10k + beliefs: p50 Δ ≤ 5 ms, p95 Δ ≤ 50 ms vs lane-off. (Generic BFS + measured +1.0 ms p50 / +35.6 ms p95; this lane is narrower.) +- **G4 — migration:** backfill shipped + doctor row (DONE in the + landing PRs); the flip release decides auto vs prompted backfill. +- **G5 — determinism/repro:** two-build byte-identity of the spine + table on a fixed corpus; ablation bench green in CI. + +## Open questions (tracked for the flip review) + +1. Production `session_id` semantics: hook-ingested beliefs chain + within a host session — chains will be long and heterogeneous. The + first G2 shadow eval should watch chain-length distribution. +2. ~~Soft-deleted/superseded beliefs~~ — resolved skip-but-continue at + traversal time (implemented; predecessor selection at write time + also keeps GC'd beliefs eligible so chains never sever). +3. Federation: spine edges are local-store only; foreign-scope beliefs + are excluded from chains (the writer runs on local inserts only). +4. ~~Ablation bench upstream~~ — done: + `benchmarks/temporal_spine_ablation.py`. + +## Historical note + +The predecessor codebase added an equivalent writer (`link_temporal`) +five days *after* its last quantified benchmark run — this mechanism had +never been benchmarked before the #1064 campaign, on either codebase. diff --git a/docs/user/CONFIG.md b/docs/user/CONFIG.md index e40a67594..2aedd4f2a 100644 --- a/docs/user/CONFIG.md +++ b/docs/user/CONFIG.md @@ -9,7 +9,7 @@ This is the reference for power users whose project has a documentation idiom or A single optional TOML file at the root of a project (or any ancestor). It exposes the following power-user surfaces: - `[noise]` — onboard-time belief filter. Changes how `aelf onboard` ingests beliefs; nothing else. -- `[retrieval]` (v1.3+) — retrieval-time tier toggles + ranking. Knobs: `entity_index_enabled` (L2.5), `bfs_enabled` (L3), `posterior_weight` (partial Bayesian-weighted L1 ranking), `l1_limit` + `token_budget` (the #1045 wide-retrieval knobs — BM25 candidate cap + token budget, default 50/2400; raise both together for multi-hop recall), `use_bm25f_anchors` (BM25F-with-anchor-text since v1.7), `use_heat_kernel` (authority scoring lane, default-on since v2.1), `use_hrr_structural` (HRR structural-query lane, default-on since v2.1), `hrr_persist` (HRR structural-index on-disk persistence, default-on since v3.0), `use_type_aware_compression` (per-belief retention-class compression, default-on since #769), `use_intentional_clustering` (co-locating related beliefs, default-on since v3.0), `expansion_gate_enabled`, `use_gamma_posterior_temperature` (default off), and `use_zeta_posterior_rerank` (default off; mutually exclusive with the γ flag — `retrieve()` raises `ValueError` when both are on). Two placeholder flags (`use_signed_laplacian`, `use_posterior_ranking`) are recognised but emit a deprecation warning if set — their lanes have not yet shipped. +- `[retrieval]` (v1.3+) — retrieval-time tier toggles + ranking. Knobs: `entity_index_enabled` (L2.5), `bfs_enabled` (L3), `posterior_weight` (partial Bayesian-weighted L1 ranking), `l1_limit` + `token_budget` (the #1045 wide-retrieval knobs — BM25 candidate cap + token budget, default 50/2400; raise both together for multi-hop recall), `use_bm25f_anchors` (BM25F-with-anchor-text since v1.7), `use_heat_kernel` (authority scoring lane, default-on since v2.1), `use_hrr_structural` (HRR structural-query lane, default-on since v2.1), `hrr_persist` (HRR structural-index on-disk persistence, default-on since v3.0), `use_type_aware_compression` (per-belief retention-class compression, default-on since #769), `use_intentional_clustering` (co-locating related beliefs, default-on since v3.0), `expansion_gate_enabled`, `use_gamma_posterior_temperature` (default off), and `use_zeta_posterior_rerank` (default off; mutually exclusive with the γ flag — `retrieve()` raises `ValueError` when both are on), `use_temporal_spine` + `temporal_spine_budget` (the #1064 chronological-adjacency lane, default off/32; pairs with `[ingest] write_temporal_spine`). Two placeholder flags (`use_signed_laplacian`, `use_posterior_ranking`) are recognised but emit a deprecation warning if set — their lanes have not yet shipped. - `[rebuilder]` (v1.4+) — context-rebuilder knobs: `turn_window_n` (default 50), `token_budget` (default 4000), `trigger_mode` (`manual`|`threshold`|`dynamic`, default `threshold`), `threshold_fraction` (default 0.6), and `query_strategy` (v1.7+, default `stack-r1-r3` since v3.0). `[rebuild_floor]` (v1.7+) sets the token-budget floors for the session-scoped and L1 belief lanes (`[rebuild_floor] session` and `[rebuild_floor] l1`). - `[onboard.llm]` (v1.3.0+) — direct-API onboard classifier gate; documented under [Keys § `[onboard.llm]`](#onboardllm-v130) below. - `[cadence]`, `[implicit_feedback]`, and `[hook_audit]` — feedback-cadence scoring, deferred retrieval-exposure feedback, and the per-turn hook audit log. Recognised here but documented in their module docstrings (`src/aelfrice/cadence.py`, `src/aelfrice/deferred_feedback.py`, `src/aelfrice/hook.py`). @@ -144,6 +144,27 @@ use_type_aware_compression = true # var overrides. use_intentional_clustering = true +# v3.9+ (#1064). Default `false` — landing posture, not the end state: +# the default-ON flip is gated on the pre-registered #1064 criteria +# (see docs/design/feature-temporal-spine.md). When true, the +# temporal-spine lane traverses TEMPORAL_NEXT chronological chains from +# the top-5 packed L1 seeds (both directions, depth 1) and appends the +# neighbours after the L1 candidates. Reaches gold that shares zero +# salient terms with the question through chronological adjacency — +# confirmed +14.6pp gold-coverage on LoCoMo, 10x its shuffled control. +# No-op on stores with zero TEMPORAL_NEXT edges (run `aelf spine +# backfill` to build the spine on an existing store, and enable the +# [ingest] write_temporal_spine writer to keep it growing). +# AELFRICE_TEMPORAL_SPINE env var overrides. +use_temporal_spine = false + +# v3.9+ (#1064). Node budget for the temporal-spine lane traversal +# (default 32). The confirmatory budget curve is monotone (~+2.5pp +# coverage per doubling at 32/64/128, no plateau) — this is the knob to +# raise for retrieval-heavy callers with the token budget to hold the +# extra candidates. AELFRICE_TEMPORAL_SPINE_BUDGET env var overrides. +temporal_spine_budget = 32 + # Placeholder flags reserved by #154 — recognised so callers can # write forward-compat config, but their lanes have not yet # shipped. Setting either to true emits a one-shot stderr @@ -152,6 +173,16 @@ use_intentional_clustering = true # use_signed_laplacian = false # use_posterior_ranking = false +[ingest] +# v3.9+ (#1064). Default `false` (same flip gate as use_temporal_spine +# above — the two flags flip together at release time but resolve +# independently). When true, every belief insert chains to its session +# predecessor with a TEMPORAL_NEXT edge (src = successor, weight 0.8), +# building the per-session temporal spine the retrieval lane traverses. +# One edge per belief, O(1) per insert. Off-path ingest is +# byte-identical. AELFRICE_TEMPORAL_SPINE_WRITE env var overrides. +write_temporal_spine = false + [rebuilder] # v3.0+ / #718 (PR #719). Selects the query-rewriting stack used by # the context rebuilder. Default `"stack-r1-r3"` since v3.0; runs @@ -467,10 +498,39 @@ Enabled by default: `compressed_beliefs` is parallel to `beliefs` (same length, Precedence (first decisive wins): env var `AELFRICE_TYPE_AWARE_COMPRESSION=0`/`1` > explicit Python kwarg `use_type_aware_compression=` > TOML `[retrieval] use_type_aware_compression` > default `true`. The default-on flip landed in #769 after the A2 + A4 bench gates (`docs/design/feature-type-aware-compression.md` §"Bench-gate / ship-or-defer policy") cleared on the lab-side `compression_a*` corpora. Composes with `use_intentional_clustering` since #878. +### `use_temporal_spine` / `temporal_spine_budget` + +v3.9+ (#1064). Default `false` / `32`. The temporal-spine retrieval lane: +an additive candidate source after L1 that traverses `TEMPORAL_NEXT` +chronological chains from the top-5 packed L1 seeds (both directions, +depth 1) and appends the neighbours — never displacing L1 pre-packing. +The mechanism is complementary to lexical matching: gold beliefs sharing +zero salient terms with the question become reachable through +chronological adjacency to beliefs that do match. No-op guard: stores +with zero `TEMPORAL_NEXT` edges get byte-identical output at ~zero cost. +Precedence: `AELFRICE_TEMPORAL_SPINE` / `AELFRICE_TEMPORAL_SPINE_BUDGET` +env → explicit kwarg → TOML → default. Default-off is the landing +posture; the default-ON flip is gated on the pre-registered criteria in +[docs/design/feature-temporal-spine.md](../design/feature-temporal-spine.md). + ### Placeholder flags `use_signed_laplacian` and `use_posterior_ranking` are reserved by #154 but their owning lanes have not yet shipped. The flags are recognised by `warn_placeholder_flags()` so writing them in `.aelfrice.toml` does not error; setting either to `true` emits a one-shot stderr deprecation warning and is otherwise a no-op. Source of truth: `PLACEHOLDER_FLAGS` in `src/aelfrice/retrieval.py`. +## `[ingest]` (v3.9+) + +### `write_temporal_spine` + +Default `false`. When enabled, every belief insert links to the previous +belief in the same session (`created_at` order, insertion-order +tie-break) with a `TEMPORAL_NEXT` edge — the per-session temporal spine +the `use_temporal_spine` retrieval lane traverses. One edge per belief, +O(1) per insert, idempotent; the off-path is byte-identical to today. +Existing stores predate the writer: `aelf spine backfill` builds their +chains (idempotent, `--dry-run` supported), and `aelf doctor` reports +spine presence + edge count. `AELFRICE_TEMPORAL_SPINE_WRITE` env var +overrides. + ## `[rebuilder]` and `[rebuild_floor]` (v1.7+) Malformed values (wrong type, out-of-range, unrecognised strategy string) in either section fall back to the field default with a `aelfrice rebuilder: ignoring …` trace to stderr. The rebuild never raises on a bad config value. diff --git a/src/aelfrice/cli.py b/src/aelfrice/cli.py index 99ccc1e9e..cf7ab0004 100644 --- a/src/aelfrice/cli.py +++ b/src/aelfrice/cli.py @@ -4510,6 +4510,21 @@ def _cmd_health(args: argparse.Namespace, out: object) -> int: for edge_type, count in sorted_entries: print(f" {edge_type:24s} {count}", file=out) # type: ignore[arg-type] print("", file=out) # type: ignore[arg-type] + # #1064 spine row: present/absent + edge count, so operators can see + # whether this store has a temporal spine to traverse (and whether + # `aelf spine backfill` is needed) without decoding edges-by-type. + n_spine = features.edges_by_type.get("TEMPORAL_NEXT", 0) + if n_spine: + print( + f"temporal spine: present ({n_spine} TEMPORAL_NEXT edges)", + file=out, # type: ignore[arg-type] + ) + else: + print( + "temporal spine: absent (run `aelf spine backfill` to build)", + file=out, # type: ignore[arg-type] + ) + print("", file=out) # type: ignore[arg-type] sentiment_state, sentiment_count = _sentiment_from_prose_state() if sentiment_state == "enabled": print( @@ -4609,6 +4624,35 @@ def _load_aelfrice_config_dict(root: Path) -> dict[str, Any] | None: return None +def _cmd_spine(args: argparse.Namespace, out: object) -> int: + """`aelf spine backfill` — build per-session TEMPORAL_NEXT chains + over the existing store (#1064). + + Existing stores predate the ingest-time spine writer; this is the + migration path (re-ingesting everything is not). Idempotent per + `(src, dst, TEMPORAL_NEXT)` triple; `--dry-run` counts what a real + run would write without touching the store. Exits 0 always — an + empty store and a fully-chained store are both no-ops, not errors. + """ + from aelfrice.temporal_spine import backfill_temporal_spine + + dry = bool(getattr(args, "dry_run", False)) + store = _open_store() + try: + report = backfill_temporal_spine(store, dry_run=dry) + finally: + store.close() + verb = "would write" if dry else "wrote" + print( + f"temporal spine backfill: {verb} {report.n_edges_written} " + f"TEMPORAL_NEXT edge(s) across {report.n_sessions} session(s) " + f"({report.n_beliefs_in_sessions} session-tagged beliefs, " + f"{report.n_edges_existing} already present)", + file=out, # type: ignore[arg-type] + ) + return 0 + + def _cmd_migrate(args: argparse.Namespace, out: object) -> int: """Copy beliefs from the legacy global DB into the active project's DB. @@ -7030,6 +7074,21 @@ def _positive_int(s: str) -> int: ) p_migrate.set_defaults(func=_cmd_migrate) + # Hidden: one-shot migration utility for stores that predate the + # #1064 temporal-spine writer (the default-ON flip release invokes + # it; manual runs are power-user only). No slash command — it's a + # migration surface, not a workflow verb. + p_spine = sub.add_parser("spine", help=argparse.SUPPRESS) + p_spine.add_argument( + "action", choices=["backfill"], + help="build per-session TEMPORAL_NEXT chains over the existing store", + ) + p_spine.add_argument( + "--dry-run", action="store_true", + help="count what backfill would write without writing", + ) + p_spine.set_defaults(func=_cmd_spine) + # Hidden: spawned by the transcript-logger hook on rotation. Manual # invocation is for batch backfill of historical JSONL — power-user only. p_ingest_transcript = sub.add_parser( diff --git a/src/aelfrice/ingest.py b/src/aelfrice/ingest.py index 14863d36d..318993897 100644 --- a/src/aelfrice/ingest.py +++ b/src/aelfrice/ingest.py @@ -315,6 +315,20 @@ def _ingest_turn_ids( if is_auto_relationship_detection_enabled(): write_semantic_edges(store, new_belief_ids=inserted) + # #1064: optionally chain this turn's new beliefs into the + # per-session temporal spine (TEMPORAL_NEXT, src = successor, + # dst = predecessor). Default-OFF (is_temporal_spine_write_enabled): + # when off, this branch is never entered and ingest is + # byte-identical to today. Gated on `inserted` so + # corroboration-only turns skip the predecessor lookups. + from aelfrice.temporal_spine import ( + is_temporal_spine_write_enabled, + write_temporal_spine, + ) + + if is_temporal_spine_write_enabled(): + write_temporal_spine(store, new_belief_ids=inserted) + return inserted diff --git a/src/aelfrice/retrieval.py b/src/aelfrice/retrieval.py index 53c0270ca..324738e5b 100644 --- a/src/aelfrice/retrieval.py +++ b/src/aelfrice/retrieval.py @@ -96,6 +96,7 @@ seeds_from_bm25, ) from aelfrice.models import ( + EDGE_TEMPORAL_NEXT, LOCK_NONE, LOCK_TIER_REFERENCE, LOCK_USER, @@ -187,6 +188,8 @@ # (resolver default False) — landing the lane + ablation only; a default # flip reverses locked #605 and is routed to a re-opened #897. HRR_EXPAND_FLAG: Final[str] = "use_hrr_expand" +TEMPORAL_SPINE_FLAG: Final[str] = "use_temporal_spine" +TEMPORAL_SPINE_BUDGET_FLAG: Final[str] = "temporal_spine_budget" # v2.1 #434 type-aware compression flag. Default-OFF at v2.0.0 until the # lab-side bench gate (A2 + A4 in docs/design/feature-type-aware-compression.md) # clears. ON populates RetrievalResult.compressed_beliefs with per-belief @@ -251,6 +254,9 @@ ENV_HRR_STRUCTURAL: Final[str] = "AELFRICE_HRR_STRUCTURAL" # #981 HRR expansion-lane env override. Tri-state like ENV_BM25F; default-OFF. ENV_HRR_EXPAND: Final[str] = "AELFRICE_HRR_EXPAND" +# #1064 temporal-spine lane flag + node-budget knob. +ENV_TEMPORAL_SPINE: Final[str] = "AELFRICE_TEMPORAL_SPINE" +ENV_TEMPORAL_SPINE_BUDGET: Final[str] = "AELFRICE_TEMPORAL_SPINE_BUDGET" # #698 HRR persist env override. "0" disables; "1" forces on. # Mirrors _ENV_PERSIST in hrr_index (same value; imported at call site). ENV_HRR_PERSIST: Final[str] = "AELFRICE_HRR_PERSIST" @@ -693,6 +699,21 @@ def _env_hrr_expand_override() -> bool | None: return None +def _env_temporal_spine_override() -> bool | None: + """Return True/False if AELFRICE_TEMPORAL_SPINE is set to a + recognised truthy/falsy value, else None. Symmetric to + `_env_hrr_expand_override`.""" + raw = os.environ.get(ENV_TEMPORAL_SPINE) + if raw is None: + return None + norm = raw.strip().lower() + if norm in _ENV_FALSY: + return False + if norm in _ENV_TRUTHY: + return True + return None + + def _env_type_aware_compression_override() -> bool | None: """Return True/False if AELFRICE_TYPE_AWARE_COMPRESSION is set to a recognised truthy/falsy value, else None. Symmetric to @@ -1806,6 +1827,68 @@ def is_hrr_expand_enabled( return False +def is_temporal_spine_enabled( + explicit: bool | None = None, + *, + start: Path | None = None, +) -> bool: + """Resolve the temporal-spine retrieval-lane flag (#1064). + + Precedence (first decisive wins): + 1. AELFRICE_TEMPORAL_SPINE env var (truthy / falsy normalised). + 2. Explicit `explicit` kwarg from the caller. + 3. `[retrieval] use_temporal_spine` in `.aelfrice.toml`. + 4. Default: **False** — the lane lands default-OFF. The default-ON + flip is gated on the pre-registered #1064 criteria (G2-G5), a + release deliverable rather than a config change. Distinct from + `AELFRICE_TEMPORAL_SPINE_WRITE` (the ingest-time writer flag in + `aelfrice.temporal_spine`) — the two flip together at release + time but resolve independently. + + Passing the flag (any rung) never raises — an unset / unrecognised + value falls through to the next rung and ultimately to False. + """ + env = _env_temporal_spine_override() + if env is not None: + return env + if explicit is not None: + return explicit + toml_value = _read_toml_flag_for(TEMPORAL_SPINE_FLAG, start) + if toml_value is not None: + return toml_value + return False + + +def resolve_temporal_spine_budget( + explicit: int | None = None, + *, + start: Path | None = None, +) -> int: + """Resolve the temporal-spine lane's node budget (#1064). + + 1. ``AELFRICE_TEMPORAL_SPINE_BUDGET`` env var (positive int). + 2. Explicit ``explicit`` kwarg from the caller. + 3. ``[retrieval] temporal_spine_budget`` in ``.aelfrice.toml``. + 4. Default: ``temporal_spine.DEFAULT_SPINE_NODE_BUDGET`` (32). + + The confirmatory budget curve is monotone (~+2.5pp coverage per + doubling at 32/64/128 with no plateau) — the effect is + budget-limited, so this is the knob the flip release revisits. + """ + from aelfrice.temporal_spine import DEFAULT_SPINE_NODE_BUDGET + + env = _env_positive_int(ENV_TEMPORAL_SPINE_BUDGET) + if env is not None: + return env + if explicit is not None: + return int(explicit) + toml_value = _read_toml_float_for(TEMPORAL_SPINE_BUDGET_FLAG, start) + return ( + int(toml_value) if toml_value is not None + else DEFAULT_SPINE_NODE_BUDGET + ) + + def is_hrr_persist_enabled( explicit: bool | None = None, *, @@ -2191,6 +2274,13 @@ class LaneTelemetry: # expansion lane merged into the candidate set (0 when the lane is off, # the default). Lets the ablation read the lane's contribution per call. hrr_expand: int = 0 + # #1064 temporal-spine lane. ``temporal_spine`` is the packed survivor + # count (what the lane actually added to the output within budget); + # ``temporal_spine_candidates`` is what the traversal discovered before + # dedup + token-budget packing. The delta is the lane's trim loss — + # the G2 flip-gate question ("does it survive the production trim"). + temporal_spine: int = 0 + temporal_spine_candidates: int = 0 # Per-process snapshot of the most recent retrieval call. Test- @@ -2963,6 +3053,9 @@ def retrieve_with_tiers( use_intentional_clustering: bool | None = None, hrr_expand_enabled: bool | None = None, hrr_struct_index_cache: HRRStructIndexCache | None = None, + temporal_spine_enabled: bool | None = None, + temporal_spine_depth: int | None = None, + temporal_spine_node_budget: int | None = None, ) -> tuple[ list[Belief], list[str], list[str], list[str], list[list[str]], ]: @@ -3172,6 +3265,58 @@ def _cost(b: Belief) -> int: seen_pre.add(b.id) used += cost + # #1064 temporal-spine lane (additive, default-OFF). Traverses + # TEMPORAL_NEXT chains from the top-5 packed L1 seeds, both + # directions, depth 1 by default, and appends the neighbours after + # the L1 candidates — never displacing them pre-packing. No-op + # guard: a store with zero TEMPORAL_NEXT edges skips the traversal + # entirely (LIMIT-1 existence probe), so spineless stores get byte- + # identical output at ~zero cost. Spine hits deliberately do NOT + # seed the BFS expansion below: the #1064 confirmatory evidence + # measured the lane as a depth-1 append-after-L1 source with BFS + # untouched, and feeding BFS is unmeasured surface (#977 keeps + # generic BFS off anyway). + temporal_spine_ids_list: list[str] = [] + n_spine_candidates = 0 + if ( + is_temporal_spine_enabled(temporal_spine_enabled) + and query.strip() + and l1_packed + and store.has_edge_type(EDGE_TEMPORAL_NEXT) + ): + from aelfrice.temporal_spine import ( + DEFAULT_SPINE_DEPTH, + DEFAULT_SPINE_SEED_COUNT, + spine_neighbors, + ) + + spine_hits = spine_neighbors( + store, + [b.id for b in l1_packed[:DEFAULT_SPINE_SEED_COUNT]], + depth=( + temporal_spine_depth if temporal_spine_depth is not None + else DEFAULT_SPINE_DEPTH + ), + node_budget=resolve_temporal_spine_budget( + temporal_spine_node_budget, + ), + ) + n_spine_candidates = len(spine_hits) + seen_spine: set[str] = ( + locked_ids | l25_ids | set(l1_ids_list) + | set(hrr_expand_ids_list) + ) + for b in spine_hits: + if b.id in seen_spine: + continue + cost = _cost(b) + if used + cost > locked_used + relevance_budget: + break + out.append(b) + temporal_spine_ids_list.append(b.id) + seen_spine.add(b.id) + used += cost + bfs_chains: list[list[str]] = [] if bfs_on and query.strip(): seeds: list[Belief] = ( @@ -3189,6 +3334,7 @@ def _cost(b: Belief) -> int: seen_ids: set[str] = ( locked_ids | l25_ids | set(l1_ids_list) | set(hrr_expand_ids_list) + | set(temporal_spine_ids_list) ) for hop in hops: if hop.belief.id in seen_ids: @@ -3211,6 +3357,8 @@ def _cost(b: Belief) -> int: expansion_gate_skipped_bfs=gate_skipped_bfs, l1_candidates=len(l1), hrr_expand=len(hrr_expand_ids_list), + temporal_spine=len(temporal_spine_ids_list), + temporal_spine_candidates=n_spine_candidates, ) return out, locked_ids_list, l25_ids_list, l1_ids_list, bfs_chains @@ -3236,6 +3384,9 @@ def retrieve_v2( use_intentional_clustering: bool | None = None, use_hrr_structural: bool | None = None, use_hrr_expand: bool | None = None, + use_temporal_spine: bool | None = None, + temporal_spine_depth: int | None = None, + temporal_spine_node_budget: int | None = None, hrr_struct_index_cache: HRRStructIndexCache | None = None, with_doc_anchors: bool = False, now_ts: int | None = None, @@ -3305,6 +3456,19 @@ def retrieve_v2( when the flag is on and no cache is passed, an ephemeral in-memory cache is built so the lane still fires (callers running the lane hot should pass an explicit cache to amortise the build). + - `use_temporal_spine` (#1064) — when True, the temporal-spine lane + runs as an additive candidate source after L1: it traverses + TEMPORAL_NEXT chains from the top-5 packed L1 seeds (both + directions, depth 1 by default, node budget 32 by default) and + appends the chronological neighbours to the candidate set. The + lane is a no-op (byte-identical output) when the store has zero + TEMPORAL_NEXT edges. Default-OFF — the default-ON flip is gated + on the pre-registered #1064 criteria. Opt in via + `AELFRICE_TEMPORAL_SPINE=1`, the kwarg, or + `[retrieval] use_temporal_spine = true`. + `temporal_spine_depth` / `temporal_spine_node_budget` tune the + traversal (budget also via `AELFRICE_TEMPORAL_SPINE_BUDGET` env + or `[retrieval] temporal_spine_budget` TOML). - `hrr_struct_index_cache` (#152) — explicit `HRRStructIndexCache` to reuse an already-built index across calls. None falls through to a fresh build per call. @@ -3382,6 +3546,9 @@ def retrieve_v2( use_intentional_clustering=use_intentional_clustering, hrr_expand_enabled=use_hrr_expand, hrr_struct_index_cache=expand_cache, + temporal_spine_enabled=use_temporal_spine, + temporal_spine_depth=temporal_spine_depth, + temporal_spine_node_budget=temporal_spine_node_budget, ) retrieve_elapsed = time.perf_counter() - retrieve_start if include_locked: diff --git a/src/aelfrice/store.py b/src/aelfrice/store.py index 93719080c..c09b7d77b 100644 --- a/src/aelfrice/store.py +++ b/src/aelfrice/store.py @@ -4115,6 +4115,20 @@ def list_contradicts_pairs(self) -> list[tuple[str, str]]: ) return [(str(r["a"]), str(r["b"])) for r in cur.fetchall()] + def has_edge_type(self, type_: str) -> bool: + """True when at least one edge of ``type_`` exists. + + LIMIT-1 existence probe — O(1) when an edge of the type exists + (the #1064 lane's common on-path), bounded by one table scan + when none does. Strictly cheaper than ``count_edges_by_type`` + (a full GROUP BY) for guards that only need presence; measured + 4.15 ms vs ~0 ms per call on a 24k-edge production store. + """ + row = self._conn.execute( + "SELECT 1 FROM edges WHERE type = ? LIMIT 1", (type_,), + ).fetchone() + return row is not None + def count_edges_by_type(self) -> dict[str, int]: """`{edge_type: count}` for every edge type in the store. @@ -4366,6 +4380,59 @@ def edges_from(self, src: str) -> list[Edge]: ) return [_row_to_edge(r) for r in cur.fetchall()] + def session_predecessor_id(self, belief_id: str) -> str | None: + """Id of the belief immediately before `belief_id` in its session. + + Ordering is `(created_at, rowid)` — creation time, with insertion + order as the tie-break for identical timestamps. Returns None when + the belief does not exist, carries a NULL `session_id`, or is the + first belief of its session. Soft-deleted (`valid_to` set) beliefs + are eligible predecessors — spine chain integrity must survive GC + (#1064; skip-but-continue happens at traversal time, not here). + + Consumer: the #1064 temporal-spine writer + (`aelfrice.temporal_spine.write_temporal_spine`) and the + `aelf spine backfill` path. Uses `idx_beliefs_session`, so the + lookup is O(log n) per call. + """ + cur = self._conn.execute( + """ + SELECT b2.id + FROM beliefs AS b1 + JOIN beliefs AS b2 + ON b2.session_id = b1.session_id + WHERE b1.id = ? + AND b1.session_id IS NOT NULL + AND (b2.created_at < b1.created_at + OR (b2.created_at = b1.created_at + AND b2.rowid < b1.rowid)) + ORDER BY b2.created_at DESC, b2.rowid DESC + LIMIT 1 + """, + (belief_id,), + ) + row = cur.fetchone() + return str(row["id"]) if row is not None else None + + def session_belief_ids_ordered(self) -> list[tuple[str, str]]: + """All `(session_id, belief_id)` pairs with a non-NULL session, + ordered by `(session_id, created_at, rowid)`. + + Consecutive rows within one session are exactly the pairs the + #1064 temporal-spine chain links. Consumer: `aelf spine backfill` + (`aelfrice.temporal_spine.backfill_temporal_spine`). + """ + cur = self._conn.execute( + """ + SELECT session_id, id + FROM beliefs + WHERE session_id IS NOT NULL + ORDER BY session_id, created_at, rowid + """ + ) + return [(str(r["session_id"]), str(r["id"])) for r in cur.fetchall()] + + def edges_from_in_scope( self, src: str, owning_scope: str | None ) -> list[Edge]: diff --git a/src/aelfrice/temporal_spine.py b/src/aelfrice/temporal_spine.py new file mode 100644 index 000000000..2d1581478 --- /dev/null +++ b/src/aelfrice/temporal_spine.py @@ -0,0 +1,408 @@ +"""Temporal spine — per-session chronological TEMPORAL_NEXT chains (#1064). + +The spine writer links each newly inserted belief to the previous belief +in the same ``session_id`` (ordered by ``created_at``, insertion order as +the tie-break) with a ``TEMPORAL_NEXT`` edge: src = the temporal +successor, dst = the predecessor (matching the ``models.py`` edge +semantics), weight ``TEMPORAL_SPINE_EDGE_WEIGHT``. One edge per belief, +O(1) per insert — the chain grows at the tail as the session grows. + +The mechanism is complementary to lexical matching: gold beliefs that +share zero salient terms with a question are unreachable by any lexical +means, but become reachable through chronological adjacency to beliefs +that *do* match. A shuffled control (identical edge count, permuted +endpoints) isolates the cause as the chronology, not the connectivity. + +Default-OFF (``is_temporal_spine_write_enabled``): a fresh install writes +no spine edges and ingest is byte-identical to today. The flag is the +landing posture, not the end state — the default-ON flip is gated on the +pre-registered criteria recorded in issue #1064 (G1–G5). Deterministic, +stdlib-only: no LLM, no embedding, no sampling (#605). + +Soft-deleted beliefs (``valid_to`` set) are not excluded from predecessor +selection: chain integrity must survive GC, so a successor links to the +most recent session predecessor regardless of its lifecycle state +(skip-but-continue happens at read time, in the traversal lane). +""" +from __future__ import annotations + +import os +import sys +import tomllib +from dataclasses import dataclass +from pathlib import Path +from typing import IO, TYPE_CHECKING, Any, Final, cast + +from aelfrice.models import EDGE_TEMPORAL_NEXT, Edge + +if TYPE_CHECKING: + from collections.abc import Sequence + + from aelfrice.models import Belief + from aelfrice.store import MemoryStore + +ENV_TEMPORAL_SPINE_WRITE: Final[str] = "AELFRICE_TEMPORAL_SPINE_WRITE" +CONFIG_FILENAME: Final[str] = ".aelfrice.toml" +SECTION: Final[str] = "ingest" +WRITE_KEY: Final[str] = "write_temporal_spine" + +# Chronological adjacency is a strong structural signal (stronger than +# the generic RELATES_TO tier) but carries no evidential content, so it +# sits below the evidential edges' 1.0. Distinct from the propagation +# valence in models.EDGE_VALENCE (0.2) — that family tunes feedback +# propagation, this is the graph-traversal edge weight. +TEMPORAL_SPINE_EDGE_WEIGHT: Final[float] = 0.8 + +_ENV_TRUTHY: Final[frozenset[str]] = frozenset({"1", "true", "yes", "on"}) +_ENV_FALSY: Final[frozenset[str]] = frozenset({"0", "false", "no", "off"}) + + +# --- Flag resolver (mirrors the #988 auto-detect resolver) -------------- + + +def _env_spine_write_override() -> bool | None: + """Return True/False if ``AELFRICE_TEMPORAL_SPINE_WRITE`` is set to a + recognised truthy / falsy value, else None (env not decisive).""" + raw = os.environ.get(ENV_TEMPORAL_SPINE_WRITE) + if raw is None: + return None + norm = raw.strip().lower() + if norm in _ENV_TRUTHY: + return True + if norm in _ENV_FALSY: + return False + return None + + +def _read_spine_write_toml(start: Path | None = None) -> bool | None: + """Read ``[ingest] write_temporal_spine`` from the nearest + `.aelfrice.toml`. Returns None on missing file / section / key / + malformed TOML / non-bool value; never raises.""" + serr: IO[str] = sys.stderr + current = (start if start is not None else Path.cwd()).resolve() + seen: set[Path] = set() + while current not in seen: + seen.add(current) + candidate = current / CONFIG_FILENAME + if candidate.is_file(): + try: + parsed: dict[str, Any] = tomllib.loads( + candidate.read_bytes().decode("utf-8", errors="replace"), + ) + except (OSError, tomllib.TOMLDecodeError) as exc: + print( + f"aelfrice temporal_spine: cannot read " + f"{WRITE_KEY} in {candidate}: {exc}", + file=serr, + ) + return None + section_obj: Any = parsed.get(SECTION, {}) + if not isinstance(section_obj, dict): + return None + val: Any = cast("dict[str, Any]", section_obj).get(WRITE_KEY) + if isinstance(val, bool): + return val + return None + if current.parent == current: + break + current = current.parent + return None + + +def is_temporal_spine_write_enabled( + explicit: bool | None = None, + *, + start: Path | None = None, +) -> bool: + """Resolve the ingest-time temporal-spine writer flag (#1064). + + Precedence (first decisive wins): + 1. ``AELFRICE_TEMPORAL_SPINE_WRITE`` env var (truthy / falsy + normalised). + 2. Explicit ``explicit`` kwarg from the caller. + 3. ``[ingest] write_temporal_spine`` in `.aelfrice.toml`. + 4. Default: False (default-OFF). + + Default-off is the landing posture: a fresh install must not start + writing spine edges at ingest until the #1064 flip-gate criteria + (G2–G5) pass. Flipping the default is that issue's deliverable 5, + not a config change. + """ + env = _env_spine_write_override() + if env is not None: + return env + if explicit is not None: + return explicit + toml_value = _read_spine_write_toml(start) + if toml_value is not None: + return toml_value + return False + + +# --- Spine writer -------------------------------------------------------- + + +@dataclass(frozen=True) +class SpineWriteReport: + """Summary of one ``write_temporal_spine`` run. + + ``n_beliefs_seen`` + Distinct belief ids processed (input order, duplicates dropped). + + ``n_edges_written`` + Beliefs that produced a new TEMPORAL_NEXT edge this run. + + ``n_skipped_no_session`` + Beliefs skipped because they carry a NULL ``session_id`` (or the + id resolved to no belief row at all) — no chain to join. + + ``n_skipped_no_predecessor`` + Beliefs that are the first of their session — nothing earlier to + link to. The next insert in the session links back to them. + + ``n_skipped_existing`` + Beliefs whose spine edge already existed (idempotency guard). + """ + + n_beliefs_seen: int + n_edges_written: int + n_skipped_no_session: int + n_skipped_no_predecessor: int + n_skipped_existing: int + + +def write_temporal_spine( + store: "MemoryStore", + *, + new_belief_ids: "Sequence[str]", +) -> SpineWriteReport: + """Chain newly inserted beliefs into their session's temporal spine. + + For each belief in ``new_belief_ids`` (this turn's inserts), find the + belief immediately before it in the same session — ordered by + ``(created_at, rowid)``, i.e. creation time with insertion order as + the tie-break — and insert one ``TEMPORAL_NEXT`` edge with + src = successor, dst = predecessor, weight + ``TEMPORAL_SPINE_EDGE_WEIGHT``. + + Per-belief work is independent (each new belief looks up its own + predecessor, which is already in the store by the time this runs), + so processing order does not affect the resulting edge set and the + output is deterministic for a given store state. + + **Idempotent.** Each ``(src, dst, TEMPORAL_NEXT)`` triple is checked + before insert and skipped if present; re-running over the same ids + writes nothing new. + + Stdlib-only: no LLM, no embedding. O(1) per belief (one indexed + predecessor lookup + one edge insert). + """ + n_seen = 0 + n_written = 0 + n_no_session = 0 + n_no_predecessor = 0 + n_existing = 0 + + processed: set[str] = set() + for belief_id in new_belief_ids: + if belief_id in processed: + continue + processed.add(belief_id) + n_seen += 1 + + belief = store.get_belief(belief_id) + if belief is None or belief.session_id is None: + n_no_session += 1 + continue + + predecessor_id = store.session_predecessor_id(belief_id) + if predecessor_id is None: + n_no_predecessor += 1 + continue + + if store.get_edge(belief_id, predecessor_id, EDGE_TEMPORAL_NEXT) is not None: + n_existing += 1 + continue + + store.insert_edge(Edge( + src=belief_id, + dst=predecessor_id, + type=EDGE_TEMPORAL_NEXT, + weight=TEMPORAL_SPINE_EDGE_WEIGHT, + )) + n_written += 1 + + return SpineWriteReport( + n_beliefs_seen=n_seen, + n_edges_written=n_written, + n_skipped_no_session=n_no_session, + n_skipped_no_predecessor=n_no_predecessor, + n_skipped_existing=n_existing, + ) + + +# --- Backfill (existing stores predate the writer) ------------------------ + + +@dataclass(frozen=True) +class SpineBackfillReport: + """Summary of one ``backfill_temporal_spine`` run. + + ``n_sessions`` + Distinct non-NULL sessions visited. + + ``n_beliefs_in_sessions`` + Beliefs carrying a session_id (chain members, including chain + heads that get no outgoing spine edge). + + ``n_edges_written`` + Consecutive pairs that produced a new TEMPORAL_NEXT edge — or, + under ``dry_run``, would have. + + ``n_edges_existing`` + Consecutive pairs whose spine edge already existed (idempotency + guard fired; re-running the backfill writes nothing new). + """ + + n_sessions: int + n_beliefs_in_sessions: int + n_edges_written: int + n_edges_existing: int + + +def backfill_temporal_spine( + store: "MemoryStore", + *, + dry_run: bool = False, +) -> SpineBackfillReport: + """Build per-session TEMPORAL_NEXT chains over an existing store. + + Existing stores predate the ingest-time writer; the migration story + cannot be "re-ingest everything". This walks every session's beliefs + in ``(created_at, rowid)`` order and links each consecutive pair + with the same edge the writer would have produced (src = successor, + dst = predecessor, weight ``TEMPORAL_SPINE_EDGE_WEIGHT``). + + **Idempotent** per ``(src, dst, TEMPORAL_NEXT)`` triple: pairs whose + edge exists are counted and skipped, so re-running after a partial + build (or on a store the writer is already chaining) is safe. + + ``dry_run`` counts what a real run would write without touching the + store. + """ + from aelfrice.models import EDGE_TEMPORAL_NEXT as _EDGE # noqa: PLC0415 + + sessions: set[str] = set() + n_beliefs = 0 + n_written = 0 + n_existing = 0 + + prev_session: str | None = None + prev_belief: str | None = None + for session_id, belief_id in store.session_belief_ids_ordered(): + sessions.add(session_id) + n_beliefs += 1 + if session_id == prev_session and prev_belief is not None: + if store.get_edge(belief_id, prev_belief, _EDGE) is not None: + n_existing += 1 + else: + if not dry_run: + store.insert_edge(Edge( + src=belief_id, + dst=prev_belief, + type=_EDGE, + weight=TEMPORAL_SPINE_EDGE_WEIGHT, + )) + n_written += 1 + prev_session = session_id + prev_belief = belief_id + + return SpineBackfillReport( + n_sessions=len(sessions), + n_beliefs_in_sessions=n_beliefs, + n_edges_written=n_written, + n_edges_existing=n_existing, + ) + + +# --- Retrieval-lane traversal --------------------------------------------- + +DEFAULT_SPINE_SEED_COUNT: Final[int] = 5 +DEFAULT_SPINE_DEPTH: Final[int] = 1 +DEFAULT_SPINE_NODE_BUDGET: Final[int] = 32 + + +def spine_neighbors( + store: "MemoryStore", + seed_ids: "Sequence[str]", + *, + depth: int = DEFAULT_SPINE_DEPTH, + node_budget: int = DEFAULT_SPINE_NODE_BUDGET, +) -> list["Belief"]: + """Chronological neighbours of ``seed_ids`` over TEMPORAL_NEXT edges. + + Bidirectional: for each frontier belief, both its temporal + successors (edges whose ``dst`` is the belief) and its predecessor + (edges whose ``src`` is the belief) are visited. Traversal is + breadth-first to ``depth`` hops, emitting at most ``node_budget`` + beliefs. The #1064 confirmatory evidence ran depth-1; the monotone + budget curve (~+2.5pp per doubling at 32/64/128) says the budget + knob is the one to revisit at flip time. + + Deterministic: seeds are processed in input order; within one + frontier belief, successors come before the predecessor and each + group is sorted by belief id. No sampling, no scores. + + Soft-deleted beliefs (``valid_to`` set) are skip-but-continue + (#1064 open question 2): they are traversed *through* — kept on the + frontier so a GC'd chain segment doesn't sever the spine — but are + never emitted and never consume ``node_budget``. + + Seeds themselves are never emitted. Returns beliefs in discovery + order (the caller packs them within its token budget). + """ + if depth <= 0 or node_budget <= 0 or not seed_ids: + return [] + + visited: set[str] = set(seed_ids) + emitted: list["Belief"] = [] + frontier: list[str] = list(dict.fromkeys(seed_ids)) + + for _hop in range(depth): + if not frontier: + break + edges = [ + e for e in store.edges_for_beliefs(list(frontier)) + if e.type == EDGE_TEMPORAL_NEXT + ] + successors: dict[str, list[str]] = {} + predecessors: dict[str, list[str]] = {} + for e in edges: + # src = temporal successor of dst (models.py semantics). + successors.setdefault(e.dst, []).append(e.src) + predecessors.setdefault(e.src, []).append(e.dst) + + next_frontier: list[str] = [] + for node in frontier: + neighbours = ( + sorted(successors.get(node, [])) + + sorted(predecessors.get(node, [])) + ) + for nid in neighbours: + if nid in visited: + continue + visited.add(nid) + belief = store.get_belief(nid) + if belief is None: + continue + if belief.valid_to is not None: + # skip-but-continue: traverse through GC'd segments. + next_frontier.append(nid) + continue + emitted.append(belief) + next_frontier.append(nid) + if len(emitted) >= node_budget: + return emitted + frontier = next_frontier + + return emitted diff --git a/tests/test_slash_commands.py b/tests/test_slash_commands.py index e252c0833..4fdf97a14 100644 --- a/tests/test_slash_commands.py +++ b/tests/test_slash_commands.py @@ -218,6 +218,10 @@ def test_upgrade_slash_keeps_step_2_imperative() -> None: # callable indefinitely as scripting / hook entry points. HIDDEN_SUBCOMMANDS = frozenset({ "statusline", "bench", "regime", "migrate", "unsetup", + # `spine` is the #1064 temporal-spine backfill (one-shot migration + # utility; the default-ON flip release invokes it). No slash + # command — it's a migration surface, not a workflow verb. + "spine", "health", "stats", "project-warm", "session-delta", "demote", "validate", "resolve", "feedback", "ingest-transcript", "sweep-feedback", diff --git a/tests/test_temporal_spine.py b/tests/test_temporal_spine.py new file mode 100644 index 000000000..8a193c143 --- /dev/null +++ b/tests/test_temporal_spine.py @@ -0,0 +1,530 @@ +"""Unit tests for the #1064 temporal-spine writer + ingest wiring. + +Covers ``write_temporal_spine`` (per-session TEMPORAL_NEXT chains, +src = successor / dst = predecessor / weight 0.8), the +``session_predecessor_id`` store accessor's ordering contract +(created_at, insertion order as tie-break), the default-off +``write_temporal_spine`` flag resolver, idempotency, and the +byte-identical off-path through ``ingest_turn``. + +All tests use a real ``MemoryStore(":memory:")`` — no mocks. +""" +from __future__ import annotations + +import hashlib + +import pytest + +from aelfrice.ingest import ingest_turn +from aelfrice.models import ( + BELIEF_FACTUAL, + EDGE_TEMPORAL_NEXT, + LOCK_NONE, + Belief, +) +from aelfrice.store import MemoryStore +from aelfrice.temporal_spine import ( + ENV_TEMPORAL_SPINE_WRITE, + TEMPORAL_SPINE_EDGE_WEIGHT, + backfill_temporal_spine, + is_temporal_spine_write_enabled, + write_temporal_spine, +) + + +def _make_belief( + store: MemoryStore, + *, + belief_id: str, + content: str, + session_id: str | None = None, + created_at: str = "2026-01-01T00:00:00Z", +) -> Belief: + b = Belief( + id=belief_id, + content=content, + content_hash=hashlib.sha256(content.encode()).hexdigest(), + alpha=1.0, + beta=1.0, + type=BELIEF_FACTUAL, + lock_level=LOCK_NONE, + locked_at=None, + created_at=created_at, + last_retrieved_at=None, + session_id=session_id, + ) + store.insert_belief(b) + return b + + +def _spine_edges(store: MemoryStore) -> list[tuple[str, str, float]]: + """All TEMPORAL_NEXT edges as (src, dst, weight), sorted.""" + rows = store._conn.execute( # type: ignore[attr-defined] + "SELECT src, dst, weight FROM edges WHERE type = ? ORDER BY src, dst", + (EDGE_TEMPORAL_NEXT,), + ).fetchall() + return [(r[0], r[1], r[2]) for r in rows] + + +@pytest.fixture +def store() -> MemoryStore: + return MemoryStore(":memory:") + + +# --------------------------------------------------------------------------- +# Flag resolver precedence +# --------------------------------------------------------------------------- + + +def test_flag_defaults_off(monkeypatch: pytest.MonkeyPatch, tmp_path) -> None: + monkeypatch.delenv(ENV_TEMPORAL_SPINE_WRITE, raising=False) + # start at an empty dir so no repo .aelfrice.toml is found + assert is_temporal_spine_write_enabled(start=tmp_path) is False + + +def test_flag_env_wins_over_kwarg(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv(ENV_TEMPORAL_SPINE_WRITE, "off") + assert is_temporal_spine_write_enabled(explicit=True) is False + monkeypatch.setenv(ENV_TEMPORAL_SPINE_WRITE, "on") + assert is_temporal_spine_write_enabled(explicit=False) is True + + +def test_flag_unrecognised_env_not_decisive( + monkeypatch: pytest.MonkeyPatch, tmp_path, +) -> None: + monkeypatch.setenv(ENV_TEMPORAL_SPINE_WRITE, "maybe") + assert is_temporal_spine_write_enabled(explicit=True, start=tmp_path) is True + assert is_temporal_spine_write_enabled(start=tmp_path) is False + + +def test_flag_kwarg_wins_over_toml( + monkeypatch: pytest.MonkeyPatch, tmp_path, +) -> None: + monkeypatch.delenv(ENV_TEMPORAL_SPINE_WRITE, raising=False) + (tmp_path / ".aelfrice.toml").write_text( + "[ingest]\nwrite_temporal_spine = true\n" + ) + assert is_temporal_spine_write_enabled(explicit=False, start=tmp_path) is False + + +def test_flag_toml_read(monkeypatch: pytest.MonkeyPatch, tmp_path) -> None: + monkeypatch.delenv(ENV_TEMPORAL_SPINE_WRITE, raising=False) + (tmp_path / ".aelfrice.toml").write_text( + "[ingest]\nwrite_temporal_spine = true\n" + ) + assert is_temporal_spine_write_enabled(start=tmp_path) is True + (tmp_path / ".aelfrice.toml").write_text( + "[ingest]\nwrite_temporal_spine = false\n" + ) + assert is_temporal_spine_write_enabled(start=tmp_path) is False + + +def test_flag_malformed_toml_not_decisive( + monkeypatch: pytest.MonkeyPatch, tmp_path, +) -> None: + monkeypatch.delenv(ENV_TEMPORAL_SPINE_WRITE, raising=False) + (tmp_path / ".aelfrice.toml").write_text( + "[ingest]\nwrite_temporal_spine = 'yes'\n" + ) + assert is_temporal_spine_write_enabled(start=tmp_path) is False + + +# --------------------------------------------------------------------------- +# session_predecessor_id ordering contract +# --------------------------------------------------------------------------- + + +def test_predecessor_orders_by_created_at(store: MemoryStore) -> None: + _make_belief(store, belief_id="b1", content="first fact", + session_id="s1", created_at="2026-01-01T00:00:01Z") + _make_belief(store, belief_id="b2", content="second fact", + session_id="s1", created_at="2026-01-01T00:00:02Z") + _make_belief(store, belief_id="b3", content="third fact", + session_id="s1", created_at="2026-01-01T00:00:03Z") + assert store.session_predecessor_id("b1") is None + assert store.session_predecessor_id("b2") == "b1" + assert store.session_predecessor_id("b3") == "b2" + + +def test_predecessor_tie_breaks_on_insertion_order(store: MemoryStore) -> None: + # Identical created_at: insertion order (rowid) decides the chain. + ts = "2026-01-01T00:00:00Z" + _make_belief(store, belief_id="z-late", content="inserted first", + session_id="s1", created_at=ts) + _make_belief(store, belief_id="a-early", content="inserted second", + session_id="s1", created_at=ts) + assert store.session_predecessor_id("z-late") is None + assert store.session_predecessor_id("a-early") == "z-late" + + +def test_predecessor_scoped_to_session(store: MemoryStore) -> None: + _make_belief(store, belief_id="b1", content="session one fact", + session_id="s1", created_at="2026-01-01T00:00:01Z") + _make_belief(store, belief_id="b2", content="session two fact", + session_id="s2", created_at="2026-01-01T00:00:02Z") + assert store.session_predecessor_id("b2") is None + + +def test_predecessor_null_session_and_missing(store: MemoryStore) -> None: + _make_belief(store, belief_id="b1", content="no session fact", + session_id=None) + assert store.session_predecessor_id("b1") is None + assert store.session_predecessor_id("nonexistent") is None + + +# --------------------------------------------------------------------------- +# write_temporal_spine +# --------------------------------------------------------------------------- + + +def test_writer_chains_session(store: MemoryStore) -> None: + _make_belief(store, belief_id="b1", content="first fact", + session_id="s1", created_at="2026-01-01T00:00:01Z") + _make_belief(store, belief_id="b2", content="second fact", + session_id="s1", created_at="2026-01-01T00:00:02Z") + + report = write_temporal_spine(store, new_belief_ids=["b1", "b2"]) + + assert report.n_beliefs_seen == 2 + assert report.n_edges_written == 1 + assert report.n_skipped_no_predecessor == 1 + assert _spine_edges(store) == [ + ("b2", "b1", TEMPORAL_SPINE_EDGE_WEIGHT), + ] + + +def test_writer_links_batch_to_prior_session_tail(store: MemoryStore) -> None: + # b1 chained in an earlier turn; a later turn's batch must link its + # first belief back to the store's existing session tail. + _make_belief(store, belief_id="b1", content="prior turn fact", + session_id="s1", created_at="2026-01-01T00:00:01Z") + write_temporal_spine(store, new_belief_ids=["b1"]) + + _make_belief(store, belief_id="b2", content="next turn fact", + session_id="s1", created_at="2026-01-01T00:00:02Z") + _make_belief(store, belief_id="b3", content="another next turn fact", + session_id="s1", created_at="2026-01-01T00:00:03Z") + report = write_temporal_spine(store, new_belief_ids=["b2", "b3"]) + + assert report.n_edges_written == 2 + assert _spine_edges(store) == [ + ("b2", "b1", TEMPORAL_SPINE_EDGE_WEIGHT), + ("b3", "b2", TEMPORAL_SPINE_EDGE_WEIGHT), + ] + + +def test_writer_sessions_isolated(store: MemoryStore) -> None: + _make_belief(store, belief_id="a1", content="session a first", + session_id="sa", created_at="2026-01-01T00:00:01Z") + _make_belief(store, belief_id="b1", content="session b first", + session_id="sb", created_at="2026-01-01T00:00:02Z") + _make_belief(store, belief_id="a2", content="session a second", + session_id="sa", created_at="2026-01-01T00:00:03Z") + + report = write_temporal_spine(store, new_belief_ids=["a1", "b1", "a2"]) + + assert report.n_edges_written == 1 + assert _spine_edges(store) == [ + ("a2", "a1", TEMPORAL_SPINE_EDGE_WEIGHT), + ] + + +def test_writer_skips_null_session(store: MemoryStore) -> None: + _make_belief(store, belief_id="b1", content="no session fact", + session_id=None) + report = write_temporal_spine(store, new_belief_ids=["b1", "ghost"]) + assert report.n_beliefs_seen == 2 + assert report.n_skipped_no_session == 2 + assert _spine_edges(store) == [] + + +def test_writer_idempotent(store: MemoryStore) -> None: + _make_belief(store, belief_id="b1", content="first fact", + session_id="s1", created_at="2026-01-01T00:00:01Z") + _make_belief(store, belief_id="b2", content="second fact", + session_id="s1", created_at="2026-01-01T00:00:02Z") + + first = write_temporal_spine(store, new_belief_ids=["b1", "b2"]) + second = write_temporal_spine(store, new_belief_ids=["b1", "b2"]) + + assert first.n_edges_written == 1 + assert second.n_edges_written == 0 + assert second.n_skipped_existing == 1 + assert len(_spine_edges(store)) == 1 + + +def test_writer_dedupes_input_ids(store: MemoryStore) -> None: + _make_belief(store, belief_id="b1", content="first fact", + session_id="s1", created_at="2026-01-01T00:00:01Z") + _make_belief(store, belief_id="b2", content="second fact", + session_id="s1", created_at="2026-01-01T00:00:02Z") + report = write_temporal_spine(store, new_belief_ids=["b2", "b2", "b1"]) + assert report.n_beliefs_seen == 2 + assert report.n_edges_written == 1 + + +# --------------------------------------------------------------------------- +# Ingest wiring +# --------------------------------------------------------------------------- + +_TURN_ONE = "The staging database runs on port 5433." +_TURN_TWO = "The staging cache was flushed after the last deploy." + + +def test_ingest_off_path_writes_no_spine_edges( + store: MemoryStore, monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv(ENV_TEMPORAL_SPINE_WRITE, raising=False) + ingest_turn(store, _TURN_ONE, "test-source", session_id="s1") + ingest_turn(store, _TURN_TWO, "test-source", session_id="s1") + assert _spine_edges(store) == [] + + +def test_ingest_on_path_chains_consecutive_turns( + store: MemoryStore, monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv(ENV_TEMPORAL_SPINE_WRITE, "1") + n1 = ingest_turn(store, _TURN_ONE, "test-source", session_id="s1", + created_at="2026-01-01T00:00:01Z") + n2 = ingest_turn(store, _TURN_TWO, "test-source", session_id="s1", + created_at="2026-01-01T00:00:02Z") + assert n1 == 1 and n2 == 1 + edges = _spine_edges(store) + assert len(edges) == 1 + src, dst, weight = edges[0] + assert weight == TEMPORAL_SPINE_EDGE_WEIGHT + # src is the later turn's belief, dst the earlier turn's belief. + src_belief = store.get_belief(src) + dst_belief = store.get_belief(dst) + assert src_belief is not None and dst_belief is not None + assert src_belief.created_at > dst_belief.created_at + + +def test_ingest_on_path_skips_other_sessions( + store: MemoryStore, monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv(ENV_TEMPORAL_SPINE_WRITE, "1") + ingest_turn(store, _TURN_ONE, "test-source", session_id="s1", + created_at="2026-01-01T00:00:01Z") + ingest_turn(store, _TURN_TWO, "test-source", session_id="s2", + created_at="2026-01-01T00:00:02Z") + assert _spine_edges(store) == [] + + +# --------------------------------------------------------------------------- +# backfill_temporal_spine +# --------------------------------------------------------------------------- + + +def _seed_two_sessions(store: MemoryStore) -> None: + _make_belief(store, belief_id="a1", content="session a first", + session_id="sa", created_at="2026-01-01T00:00:01Z") + _make_belief(store, belief_id="a2", content="session a second", + session_id="sa", created_at="2026-01-01T00:00:02Z") + _make_belief(store, belief_id="a3", content="session a third", + session_id="sa", created_at="2026-01-01T00:00:03Z") + _make_belief(store, belief_id="b1", content="session b first", + session_id="sb", created_at="2026-01-01T00:00:02Z") + _make_belief(store, belief_id="b2", content="session b second", + session_id="sb", created_at="2026-01-01T00:00:04Z") + _make_belief(store, belief_id="n1", content="no session fact", + session_id=None) + + +def test_backfill_chains_all_sessions(store: MemoryStore) -> None: + _seed_two_sessions(store) + report = backfill_temporal_spine(store) + assert report.n_sessions == 2 + assert report.n_beliefs_in_sessions == 5 + assert report.n_edges_written == 3 + assert report.n_edges_existing == 0 + assert _spine_edges(store) == [ + ("a2", "a1", TEMPORAL_SPINE_EDGE_WEIGHT), + ("a3", "a2", TEMPORAL_SPINE_EDGE_WEIGHT), + ("b2", "b1", TEMPORAL_SPINE_EDGE_WEIGHT), + ] + + +def test_backfill_dry_run_writes_nothing(store: MemoryStore) -> None: + _seed_two_sessions(store) + report = backfill_temporal_spine(store, dry_run=True) + assert report.n_edges_written == 3 + assert _spine_edges(store) == [] + + +def test_backfill_idempotent(store: MemoryStore) -> None: + _seed_two_sessions(store) + first = backfill_temporal_spine(store) + second = backfill_temporal_spine(store) + assert first.n_edges_written == 3 + assert second.n_edges_written == 0 + assert second.n_edges_existing == 3 + assert len(_spine_edges(store)) == 3 + + +def test_backfill_matches_writer_output(store: MemoryStore) -> None: + """A backfilled store and a writer-chained store produce the same + spine — the migration path and the ingest path are equivalent.""" + _seed_two_sessions(store) + incremental = MemoryStore(":memory:") + _seed_two_sessions(incremental) + for bid in ("a1", "a2", "a3", "b1", "b2", "n1"): + write_temporal_spine(incremental, new_belief_ids=[bid]) + + backfill_temporal_spine(store) + assert _spine_edges(store) == _spine_edges(incremental) + + +def test_backfill_empty_store(store: MemoryStore) -> None: + report = backfill_temporal_spine(store) + assert report.n_sessions == 0 + assert report.n_beliefs_in_sessions == 0 + assert report.n_edges_written == 0 + + +# --------------------------------------------------------------------------- +# spine_neighbors traversal +# --------------------------------------------------------------------------- + +from aelfrice.temporal_spine import spine_neighbors # noqa: E402 + + +def _chain(store: MemoryStore, ids: list[str], *, session: str = "s1") -> None: + for i, bid in enumerate(ids): + _make_belief(store, belief_id=bid, content=f"unique fact number {bid}", + session_id=session, + created_at=f"2026-01-01T00:00:{i:02d}Z") + backfill_temporal_spine(store) + + +def test_neighbors_bidirectional_depth_one(store: MemoryStore) -> None: + _chain(store, ["b1", "b2", "b3"]) + hits = spine_neighbors(store, ["b2"]) + assert [b.id for b in hits] == ["b3", "b1"] # successor first, then pred + + +def test_neighbors_depth_two(store: MemoryStore) -> None: + _chain(store, ["b1", "b2", "b3", "b4", "b5"]) + hits = spine_neighbors(store, ["b3"], depth=2) + assert {b.id for b in hits} == {"b1", "b2", "b4", "b5"} + + +def test_neighbors_budget_caps_output(store: MemoryStore) -> None: + _chain(store, ["b1", "b2", "b3", "b4", "b5"]) + hits = spine_neighbors(store, ["b3"], depth=2, node_budget=2) + assert len(hits) == 2 + assert spine_neighbors(store, ["b3"], node_budget=0) == [] + + +def test_neighbors_skip_but_continue_soft_deleted(store: MemoryStore) -> None: + _chain(store, ["b1", "b2", "b3"]) + store.soft_delete_belief("b2") + hits = spine_neighbors(store, ["b1"], depth=2) + # b2 is traversed through (chain integrity) but never emitted. + assert [b.id for b in hits] == ["b3"] + + +def test_neighbors_seeds_never_emitted(store: MemoryStore) -> None: + _chain(store, ["b1", "b2"]) + hits = spine_neighbors(store, ["b1", "b2"], depth=3) + assert hits == [] + + +# --------------------------------------------------------------------------- +# Retrieval lane (retrieve_v2 wiring) +# --------------------------------------------------------------------------- + +from aelfrice.retrieval import ( # noqa: E402 + ENV_TEMPORAL_SPINE, + is_temporal_spine_enabled, + last_lane_telemetry, + resolve_temporal_spine_budget, + retrieve_v2, +) + +# The query shares terms with the anchor belief only; the chronological +# neighbours are lexically disjoint from it (the #1064 mechanism: ~84% +# of missing gold shares zero salient terms with the question). +_ANCHOR = "the kubernetes deployment rollout failed during the canary stage" +_BEFORE = "morning standup covered vacation plans and a birthday cake" +_AFTER = "someone watered the office plants and refilled the coffee pot" +_QUERY = "kubernetes canary rollout failure" + + +def _seed_lane_store(store: MemoryStore) -> None: + _make_belief(store, belief_id="before", content=_BEFORE, + session_id="s1", created_at="2026-01-01T00:00:01Z") + _make_belief(store, belief_id="anchor", content=_ANCHOR, + session_id="s1", created_at="2026-01-01T00:00:02Z") + _make_belief(store, belief_id="after", content=_AFTER, + session_id="s1", created_at="2026-01-01T00:00:03Z") + backfill_temporal_spine(store) + + +def test_lane_default_off_flag() -> None: + assert is_temporal_spine_enabled() is False + assert is_temporal_spine_enabled(explicit=True) is True + assert resolve_temporal_spine_budget() == 32 + assert resolve_temporal_spine_budget(explicit=7) == 7 + + +def test_lane_off_omits_neighbours(store: MemoryStore) -> None: + _seed_lane_store(store) + result = retrieve_v2(store, _QUERY, use_temporal_spine=False) + ids = {b.id for b in result.beliefs} + assert "anchor" in ids + assert "before" not in ids and "after" not in ids + tel = last_lane_telemetry() + assert tel.temporal_spine == 0 + assert tel.temporal_spine_candidates == 0 + + +def test_lane_on_appends_chronological_neighbours(store: MemoryStore) -> None: + _seed_lane_store(store) + result = retrieve_v2(store, _QUERY, use_temporal_spine=True) + ids = [b.id for b in result.beliefs] + assert "anchor" in ids + assert "before" in ids and "after" in ids + # Never displaces L1 pre-packing: neighbours come after the anchor. + assert ids.index("anchor") < ids.index("before") + assert ids.index("anchor") < ids.index("after") + tel = last_lane_telemetry() + assert tel.temporal_spine == 2 + assert tel.temporal_spine_candidates == 2 + + +def test_lane_env_var_enables( + store: MemoryStore, monkeypatch: pytest.MonkeyPatch, +) -> None: + _seed_lane_store(store) + monkeypatch.setenv(ENV_TEMPORAL_SPINE, "1") + result = retrieve_v2(store, _QUERY) + ids = {b.id for b in result.beliefs} + assert "before" in ids and "after" in ids + + +def test_lane_noop_guard_without_spine_edges(store: MemoryStore) -> None: + # Same store shape but NO spine edges: lane on must be byte-identical + # to lane off (the no-op guard for spineless stores). + _make_belief(store, belief_id="anchor", content=_ANCHOR, + session_id="s1", created_at="2026-01-01T00:00:02Z") + _make_belief(store, belief_id="other", content=_BEFORE, + session_id="s1", created_at="2026-01-01T00:00:01Z") + off = retrieve_v2(store, _QUERY, use_temporal_spine=False) + on = retrieve_v2(store, _QUERY, use_temporal_spine=True) + assert [b.id for b in on.beliefs] == [b.id for b in off.beliefs] + tel = last_lane_telemetry() + assert tel.temporal_spine == 0 + + +def test_lane_node_budget_kwarg(store: MemoryStore) -> None: + _seed_lane_store(store) + result = retrieve_v2( + store, _QUERY, use_temporal_spine=True, + temporal_spine_node_budget=1, + ) + ids = {b.id for b in result.beliefs} + # Budget 1 → exactly one neighbour emitted by the traversal. + assert len(ids & {"before", "after"}) == 1 + tel = last_lane_telemetry() + assert tel.temporal_spine_candidates == 1