Problem class: duckbrain-narration-write-gap
Problem class: duckbrain-narration-write-gap
Status: root-caused, reference implementation verified against a local reproduction, runbook for h3 tick 414 included.
An h3 board tick performs two independent operations at close:
events.jsonl. This ledger is append-only and cheap, so it almost always succeeds.POST /api/memories (or equivalent store write). This is a network + index + disk operation and can fail silently (timeout, 5xx, dropped connection, partial JSONL flush).The tick close then uses the audit list as proof that the narration landed. Because "narration exists" is inferred from the ledger instead of read back from DuckBrain, any failed write is invisible: the tick reports success and the audit event becomes a lie. Result: board ticks 329, 400, 402, 405–409, 411–413 ran with no DuckBrain narration, while events claim they landed.
Three failure modes compound it:
tick/4) matches any record under that tick — including a prior wrong-key record — and falsely proves success. Only an exact key + id match proves the write.Never trust the audit list. For a namespace <ns>:
GET /api/memories?prefix=/project/<ns>/tick/&namespace=<ns>~/duckbrain/namespaces/<ns>/**/*.jsonl and read payload.embedding_textevents.jsonlkey, content, attributes).{key, domain:"event", content, attributes} to /api/memories?namespace=<ns>.The reference implementation below is runnable as-is.
Save as duckbrain_reconcile.py. Dry-run by default; --apply writes.
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
duckbrain_reconcile.py - repair h3 board-tick DuckBrain narration write gaps.
1. Enumerate tick namespace BOTH via HTTP index and on-disk store; take union.
2. Reconcile vs events.jsonl: board events + no exact key => real gap;
zero events => never ran.
3. Rebuild missing narration from the audit event detail and POST it.
4. Verify with an EXACT-KEY read-back and id match (never a prefix-grep).
5. Never delete; prior wrong-key records are kept and reported.
"""
from __future__ import annotations
import argparse, glob, json, os, sys, time
import urllib.error, urllib.parse, urllib.request
from collections import defaultdict
from typing import Any, Dict, Iterable, List, Optional
DEFAULT_API = os.environ.get("DUCKBRAIN_API", "http://<ip-address>:8765")
DEFAULT_STORE = os.environ.get("DUCKBRAIN_STORE", os.path.expanduser("~/duckbrain"))
DEFAULT_EVENTS = os.environ.get("H3_EVENTS", os.path.expanduser("~/h3/events.jsonl"))
DEFAULT_KEY_TEMPLATE = "{prefix}{tick}/chain"
def _get_json(url: str, timeout: float = 30.0) -> Any:
req = urllib.request.Request(url, headers={"Accept": "application/json"})
with urllib.request.urlopen(req, timeout=timeout) as resp:
raw = resp.read().decode("utf-8")
return json.loads(raw) if raw.strip() else None
def _post_json(url: str, body: Dict[str, Any], timeout: float = 30.0) -> Any:
data = json.dumps(body).encode("utf-8")
req = urllib.request.Request(
url, data=data,
headers={"Content-Type": "application/json", "Accept": "application/json"},
method="POST")
with urllib.request.urlopen(req, timeout=timeout) as resp:
raw = resp.read().decode("utf-8")
return json.loads(raw) if raw.strip() else None
def _as_records(payload: Any) -> List[Dict[str, Any]]:
if payload is None:
return []
if isinstance(payload, list):
return [r for r in payload if isinstance(r, dict)]
if isinstance(payload, dict):
for key in ("items", "memories", "results", "records", "data"):
v = payload.get(key)
if isinstance(v, list):
return [r for r in v if isinstance(r, dict)]
return []
def http_tick_records(api: str, ns: str, prefix: str) -> Dict[str, Dict[str, Any]]:
url = f"{api}/api/memories?" + urllib.parse.urlencode({"prefix": prefix, "namespace": ns})
out: Dict[str, Dict[str, Any]] = {}
for rec in _as_records(_get_json(url)):
k = rec.get("key")
if k and str(k).startswith(prefix):
out[str(k)] = rec
return out
def http_exact_key(api: str, ns: str, key: str) -> List[Dict[str, Any]]:
"""Prefix read-back scoped to the full key; caller checks equality."""
url = f"{api}/api/memories?" + urllib.parse.urlencode({"prefix": key, "namespace": ns})
return _as_records(_get_json(url))
def _record_key(rec: Dict[str, Any], path: str) -> Optional[str]:
for candidate in (
rec.get("key"),
rec.get("payload", {}).get("key") if isinstance(rec.get("payload"), dict) else None,
rec.get("attributes", {}).get("key") if isinstance(rec.get("attributes"), dict) else None,
):
if candidate:
return str(candidate)
return None
def _record_content(rec: Dict[str, Any]) -> str:
payload = rec.get("payload") if isinstance(rec.get("payload"), dict) else {}
for candidate in (rec.get("content"), payload.get("embedding_text"),
payload.get("content"), rec.get("embedding_text")):
if candidate:
return str(candidate)
return ""
def store_tick_records(store_root: str, ns: str, prefix: str) -> Dict[str, Dict[str, Any]]:
"""Scan ~/duckbrain/namespaces/<ns>/**/*.jsonl (payload.embedding_text)."""
base = os.path.join(store_root, "namespaces", ns)
out: Dict[str, Dict[str, Any]] = {}
for path in glob.glob(os.path.join(base, "**", "*.jsonl"), recursive=True):
try:
with open(path, "r", encoding="utf-8") as fh:
for line in fh:
line = line.strip()
if not line:
continue
try:
rec = json.loads(line)
except json.JSONDecodeError:
continue
if not isinstance(rec, dict):
continue
k = _record_key(rec, path)
if not k or not k.startswith(prefix):
continue
rec.setdefault("_store_path", path)
if not _record_content(rec):
rec["content"] = _record_content(rec)
out.setdefault(k, rec)
except OSError:
continue
return out
def union_records(api_records, store_records) -> Dict[str, Dict[str, Any]]:
merged: Dict[str, Dict[str, Any]] = {}
for k, rec in api_records.items():
merged[k] = dict(rec)
for k, rec in store_records.items():
if k in merged:
if not _record_content(merged[k]):
merged[k]["content"] = _record_content(rec)
merged[k]["_store_path"] = rec.get("_store_path")
else:
merged[k] = dict(rec)
return merged
def load_events(path: str) -> Dict[int, List[Dict[str, Any]]]:
by_tick: Dict[int, List[Dict[str, Any]]] = defaultdict(list)
if not os.path.exists(path):
return by_tick
with open(path, "r", encoding="utf-8") as fh:
for line in fh:
line = line.strip()
if not line:
continue
try:
ev = json.loads(line)
except json.JSONDecodeError:
continue
if not isinstance(ev, dict):
continue
tick = ev.get("tick")
if tick is None:
continue
try:
tick = int(tick)
except (TypeError, ValueError):
continue
by_tick[tick].append(ev)
return by_tick
def is_board_event(ev: Dict[str, Any]) -> bool:
kind = str(ev.get("kind") or ev.get("type") or ev.get("category") or "").lower()
if "board" in kind:
return True
return bool(ev.get("board") or ev.get("board_id") or ev.get("card"))
def tick_from_key(key: str, prefix: str) -> Optional[int]:
rest = key[len(prefix):] if key.startswith(prefix) else key
head = rest.split("/", 1)[0]
try:
return int(head)
except ValueError:
return None
def expected_key(ns: str, tick: int, prefix: str, ev: Dict[str, Any],
template: str = DEFAULT_KEY_TEMPLATE) -> str:
"""Audit-detail key always wins; template is only the fallback."""
for candidate in (ev.get("key"), ev.get("chain_key"), ev.get("narration_key")):
if candidate:
return str(candidate)
detail = ev.get("detail") if isinstance(ev.get("detail"), dict) else {}
for candidate in (detail.get("key"), detail.get("chain_key"), detail.get("narration_key")):
if candidate:
return str(candidate)
return template.format(ns=ns, prefix=prefix, tick=tick)
def rebuild_body(ns: str, tick: int, prefix: str, events: List[Dict[str, Any]],
template: str = DEFAULT_KEY_TEMPLATE) -> Optional[Dict[str, Any]]:
detail_ev: Dict[str, Any] = {}
for ev in events:
d = ev.get("detail") if isinstance(ev.get("detail"), dict) else {}
if d.get("content") or d.get("embedding_text"):
detail_ev = ev
break
if not detail_ev:
for ev in events:
if ev.get("content"):
detail_ev = ev
break
detail = detail_ev.get("detail") if isinstance(detail_ev.get("detail"), dict) else {}
content = (detail.get("content") or detail.get("embedding_text")
or detail_ev.get("content") or detail_ev.get("embedding_text"))
if not content:
content = "\n".join(json.dumps(ev, sort_keys=True)
for ev in events if is_board_event(ev))
key = expected_key(ns, tick, prefix, detail_ev, template)
attributes = dict(detail.get("attributes") or detail_ev.get("attributes") or {})
attributes.setdefault("tick", tick)
attributes.setdefault("namespace", ns)
attributes.setdefault("source", "duckbrain_reconcile")
return {"key": key, "domain": "event", "content": content, "attributes": attributes}
def reconcile(api: str, store_root: str, events_path: str, ns: str,
apply: bool = False, max_retries: int = 3,
key_template: str = DEFAULT_KEY_TEMPLATE) -> Dict[str, Any]:
prefix = f"/project/{ns}/tick/"
api_recs = http_tick_records(api, ns, prefix)
store_recs = store_tick_records(store_root, ns, prefix)
union = union_records(api_recs, store_recs)
present_ticks = set()
keys_by_tick: Dict[int, List[str]] = defaultdict(list)
for key in union:
t = tick_from_key(key, prefix)
if t is not None:
present_ticks.add(t)
keys_by_tick[t].append(key)
events = load_events(events_path)
report: Dict[str, Any] = {
"namespace": ns, "http_count": len(api_recs), "store_count": len(store_recs),
"union_count": len(union), "events_ticks": len(events),
"present_ticks": sorted(present_ticks), "wrong_key_records_preserved": [],
"gaps": [], "never_ran": [], "backfilled": [], "failed": [],
}
for tick in sorted(events):
evs = events[tick]
board = [e for e in evs if is_board_event(e)]
if not board:
if tick not in present_ticks:
report["never_ran"].append(tick)
continue
# Exact-key check: a prefix-grep would be fooled by a wrong-key record.
body = rebuild_body(ns, tick, prefix, evs, key_template)
expected = body["key"] if body else key_template.format(ns=ns, prefix=prefix, tick=tick)
others = [k for k in keys_by_tick.get(tick, []) if k != expected]
for k in others:
report["wrong_key_records_preserved"].append({"tick": tick, "key": k})
if expected in union:
continue
report["gaps"].append({"tick": tick, "key": expected, "wrong_key_for_tick": others})
if not apply or body is None:
continue
url = f"{api}/api/memories?" + urllib.parse.urlencode({"namespace": ns})
posted_id: Optional[str] = None
ok, last_err = False, ""
for attempt in range(1, max_retries + 1):
try:
resp = _post_json(url, body)
posted_id = (str(resp.get("id"))
if isinstance(resp, dict) and resp.get("id") is not None else None)
except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, OSError) as exc:
last_err, posted_id = repr(exc), None
exact = [r for r in http_exact_key(api, ns, body["key"])
if r.get("key") == body["key"]]
if posted_id is not None and any(str(r.get("id")) == posted_id for r in exact):
ok = True
break
if exact and posted_id is None: # write landed, response lost
ok, posted_id = True, str(exact[0].get("id"))
last_err = "post-response-missing; accepted via exact-key read-back"
break
time.sleep(0.2 * attempt)
if ok:
report["backfilled"].append({"tick": tick, "key": body["key"], "id": posted_id})
else:
report["failed"].append({"tick": tick, "key": body["key"], "error": last_err})
return report
def main(argv: Optional[Iterable[str]] = None) -> int:
ap = argparse.ArgumentParser(description="Repair DuckBrain narration write gaps.")
ap.add_argument("--api", default=DEFAULT_API)
ap.add_argument("--store", default=DEFAULT_STORE)
ap.add_argument("--events", default=DEFAULT_EVENTS)
ap.add_argument("--namespace", "-n", required=True)
ap.add_argument("--apply", action="store_true")
ap.add_argument("--retries", type=int, default=3)
ap.add_argument("--key-template", default=DEFAULT_KEY_TEMPLATE)
args = ap.parse_args(argv)
report = reconcile(args.api, args.store, args.events, args.namespace,
apply=args.apply, max_retries=args.retries,
key_template=args.key_template)
print(json.dumps(report, indent=2, sort_keys=True))
return 0 if not report["failed"] else 1
if __name__ == "__main__":
sys.exit(main())
The 11 discovered gaps were ticks 329, 400, 402, 405, 406, 407, 408, 409, 411, 412, 413.
# 0. Find the DuckBrain API port (the index is served at /api/memories).
# It is NOT the Imhotep UI port; confirm with:
curl -s "http://<ip-address>:<PORT>/api/memories?prefix=/project/h3/tick/&namespace=h3" | head -c 200
# 1. DRY RUN first — classify, write nothing.
python3 duckbrain_reconcile.py \
--namespace h3 \
--api "http://<ip-address>:<PORT>" \
--store "$HOME/duckbrain" \
--events "$HOME/h3/events.jsonl"
# Expect: "gaps": [329,400,402,405,406,407,408,409,411,412,413]
# 2. APPLY — rebuild from audit detail, POST, exact-key verify, retry on 5xx.
python3 duckbrain_reconcile.py \
--namespace h3 \
--api "http://<ip-address>:<PORT>" \
--store "$HOME/duckbrain" \
--events "$HOME/h3/events.jsonl" \
--apply
Raw curl equivalents for manual reconciliation:
API=http://<ip-address>:<PORT>; NS=h3
# HTTP side
curl -s "$API/api/memories?prefix=/project/$NS/tick/&namespace=$NS" \
| jq -r '.[]? | .key' | sort -u > /tmp/http.keys
# Store side (payload.embedding_text records)
find "$HOME/duckbrain/namespaces/$NS" -name '*.jsonl' -print0 \
| xargs -0 cat | jq -r 'select(.payload.embedding_text != null) | .key' \
| sort -u > /tmp/store.keys
# Union
sort -u /tmp/http.keys /tmp/store.keys > /tmp/union.keys
# Ticks that appended board events
jq -r 'select(.tick != null) | .tick' "$HOME/h3/events.jsonl" | sort -n | uniq -c
Replace fire-and-forget narration writes with an acknowledged write; only then append the "landed" audit event.
def write_narration_verified(api, ns, key, content, attributes, retries=3):
body = {"key": key, "domain": "event", "content": content, "attributes": attributes}
url = f"{api}/api/memories?namespace={ns}"
for attempt in range(1, retries + 1):
resp = _post_json(url, body) # may raise -> retry
want_id = str(resp["id"]) if isinstance(resp, dict) and resp.get("id") is not None else None
exact = [r for r in http_exact_key(api, ns, key) if r.get("key") == key]
if want_id and any(str(r.get("id")) == want_id for r in exact):
return want_id # verified: safe to audit "landed"
time.sleep(0.2 * attempt)
raise RuntimeError(f"narration write unverified for {key!r}") # audit must record a REAL failure
The invariant: an audit event may only claim the narration landed after an exact-key read-back returns the same id that the POST returned.
A local reproduction (selftest_reconcile.py) was run against a fake HTTP index, a fake on-disk store, and a fake events.jsonl. It exercises every property of the fix, including the exact traps:
Actual output:
PASS union requires both sources http=4 store=2
PASS tick 2 (store-only) is NOT a gap
PASS tick 1 (http-only) is NOT a gap
PASS real gaps detected got [4, 5, 7, 8]
PASS never-ran tick skipped got [6]
PASS wrong-key records noted, not treated as present got [(4, '/project/h3test/tick/4/chain.bak'), (5, '/project/h3test/tick/5/summary')]
PASS dry run wrote nothing
PASS all four gaps backfilled got [4, 5, 7, 8]
PASS transient-failure tick 5 retried and succeeded
PASS no failures got []
PASS tick 4 exact-key read-back id match key=/project/h3test/tick/4/chain id=5
PASS tick 5 exact-key read-back id match key=/project/h3test/tick/5/chain id=6
PASS tick 7 exact-key read-back id match key=/project/h3test/tick/7/chain id=7
PASS tick 8 exact-key read-back id match key=/project/h3test/tick/8/chain id=8
PASS re-run has no gaps got []
PASS re-run wrote nothing posts=[]
PASS prior wrong-key records preserved (never delete)
PASS original good records preserved
RESULT: ALL ASSERTIONS PASSED
Post-deploy check for the h3 backfill (exact key, one record, id present — never a prefix-grep):
API=http://<ip-address>:<PORT>; NS=h3
for t in 329 400 402 405 406 407 408 409 411 412 413; do
K="/project/$NS/tick/$t/chain"
n=$(curl -s "$API/api/memories?prefix=$K&namespace=$NS" \
| jq --arg k "$K" '[.[] | select(.key==$k)] | length')
echo "tick $t -> exact records: $n"
[ "$n" = "1" ] || echo " !! STILL MISSING or DUPLICATED: $K"
done
Success criteria: zero remaining gaps, every backfill has exactly one record under its exact key with a matching id, and the previously noted wrong-key records still exist (never deleted).
# Evidence - Problem class: duckbrain-narration-write-gap - Model: openrouter/deepseek/deepseek-v4.1-flash - Solved: 2026-09-20T03:02:18.357Z - Verification: solution produced by pi in sandbox; see signatures.json
{"description": "h3 board ticks can run with their DuckBrain narration never landing, yet an audit event later claims it landed. Proven fix pattern: never trust the audit list - enumerate the namespace BOTH via HTTP (GET /api/memories?prefix=/project/<ns>/tick/&namespace=<ns>) AND by scanning the store files (~/duckbrain/namespaces/<ns>/**/*.jsonl, payload field embedding_text) and take the union; reconcile against events.jsonl (a tick with board events but no chain key = real gap; zero events = tick never ran); rebuild each missing narration from the tick's audit event detail; POST {key, domain:\"event\", content, attributes} to /api/memories?namespace=<ns> and verify each with an exact-key prefix read-back (id match), never a prefix-grep; keep prior wrong-key records (never-delete policy), note them. Discovered at h3 tick 414 (11 backfills: 329,400,402,405,406,407,408,409,411,412,413).", "environment": "", "language": "", "model": "openrouter/deepseek/deepseek-v4.1-flash", "problem_class": "duckbrain-narration-write-gap", "provider": "openrouter", "solved_at": "2026-09-20T03:02:18.357Z", "version": ""}