Agent history Python API

The readers and the schema contract of habemus-papadum-agent-history. The TypeScript side is documented in the TypeScript API reference.

The DuckDB schema shared by every ingester and by the TypeScript query package.

Every table carries source (codex-fs, claude-fs, codex-api, live) and origin (the concrete input such as state_5, rollout or thread/items/list). Timestamps are naive UTC. Blobs live in their own tables so a later MotherDuck migration can move them to object storage without touching the fact tables. fork_edges records which thread inherited context from which (continued sessions, forks, forked sub-agents) and threads.lineage_id names the root of that chain, which is what a person means by "a session". thread_links holds the per-thread facts those are derived from and ingest_ledger what each source's files looked like when they were last read, so a run can replace only the threads whose files changed. :func:describe renders the contract as JSON; the TypeScript package keeps a copy and a parity test fails when they drift.

SCHEMA_VERSION = 5 module-attribute

Bump when a table, column or view changes shape; the TypeScript parity test reads it.

VIEWS = {'usage_events': "\nSELECT source_id, thread_id, turn_id, ts, rollout_ordinal, model, input_tokens, cached_input_tokens,\n cache_write_input_tokens, cache_write_1h_input_tokens, output_tokens, reasoning_output_tokens,\n wasted_output_tokens, total_tokens, 'usage_record' AS basis, source\nFROM (SELECT * FROM responses WHERE origin <> 'copied-context'\n QUALIFY row_number() OVER (PARTITION BY source, response_id ORDER BY ts, rollout_ordinal) = 1)\nUNION ALL\nSELECT source_id, thread_id, turn_id, ts, rollout_ordinal, NULL AS model,\n total_input_tokens - prev_input, total_cached_input_tokens - prev_cached,\n total_cache_write_input_tokens - prev_cache_write, NULL AS cache_write_1h_input_tokens,\n total_output_tokens - prev_output, total_reasoning_output_tokens - prev_reasoning, NULL AS wasted_output_tokens,\n total_tokens - prev_total, 'counter_delta' AS basis, source\nFROM (SELECT *,\n lag(total_input_tokens, 1, 0) OVER w AS prev_input,\n lag(total_cached_input_tokens, 1, 0) OVER w AS prev_cached,\n lag(COALESCE(total_cache_write_input_tokens, 0), 1, 0) OVER w AS prev_cache_write,\n lag(total_output_tokens, 1, 0) OVER w AS prev_output,\n lag(total_reasoning_output_tokens, 1, 0) OVER w AS prev_reasoning,\n lag(total_tokens, 1, 0) OVER w AS prev_total\n FROM token_counts\n WHERE total_tokens IS NOT NULL AND thread_id NOT IN (SELECT DISTINCT thread_id FROM responses)\n WINDOW w AS (PARTITION BY thread_id ORDER BY rollout_ordinal))\nWHERE total_tokens - prev_total > 0\n"} module-attribute

Per-response usage where recorded, else positive deltas of cumulative counters.

Threads with usage records never enter the delta branch, so nothing is counted twice, and responses a file copied from another session or parent (origin = 'copied-context') are left out. The first counter value of a thread counts as an increment from zero; a fork inherits its parent's counters, so that first increment can include inherited history.

create(connection)

Create every table and view when absent.

Parameters:

Name	Type	Description	Default
connection	DuckDBPyConnection	Open connection to the target database.	required

Source code in packages/agent-history/src/pdum/agent_history/schema.py

def create(connection: Any) -> None:
    """Create every table and view when absent.

    Parameters
    ----------
    connection : duckdb.DuckDBPyConnection
        Open connection to the target database.
    """
    for statement in ddl():
        connection.execute(statement)

ddl()

Return the CREATE statements for every table and view, in dependency order.

Source code in packages/agent-history/src/pdum/agent_history/schema.py

def ddl() -> list[str]:
    """Return the CREATE statements for every table and view, in dependency order."""
    statements = []
    for table, columns in COLUMNS.items():
        rendered = ", ".join(f"{name} {kind}" for name, kind in columns)
        statements.append(f"CREATE TABLE IF NOT EXISTS {table} ({rendered})")
    for view, body in VIEWS.items():
        statements.append(f"CREATE OR REPLACE VIEW {view} AS {body.strip()}")
    return statements

describe()

Render the schema contract as JSON-compatible data.

Returns:

Type	Description
dict	{"version": int, "tables": {name: [[column, type], ...]}, "views": {name: sql}}.

Source code in packages/agent-history/src/pdum/agent_history/schema.py

def describe() -> dict[str, Any]:
    """Render the schema contract as JSON-compatible data.

    Returns
    -------
    dict
        ``{"version": int, "tables": {name: [[column, type], ...]}, "views": {name: sql}}``.
    """
    return {
        "version": SCHEMA_VERSION,
        "tables": {table: [list(column) for column in columns] for table, columns in COLUMNS.items()},
        "views": {view: " ".join(body.split()) for view, body in VIEWS.items()},
    }

describe_json()

Return :func:describe as stable, indented JSON text.

Source code in packages/agent-history/src/pdum/agent_history/schema.py

def describe_json() -> str:
    """Return :func:`describe` as stable, indented JSON text."""
    return json.dumps(describe(), indent=2, sort_keys=True) + "\n"

from_iso(value)

Convert ISO-8601 text such as 2026-10-06T13:30:29.834Z to naive UTC.

Source code in packages/agent-history/src/pdum/agent_history/schema.py

def from_iso(value: Any) -> datetime | None:
    """Convert ISO-8601 text such as ``2026-10-06T13:30:29.834Z`` to naive UTC."""
    if not isinstance(value, str) or not value:
        return None
    try:
        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError:
        return None
    if parsed.tzinfo is not None:
        parsed = parsed.astimezone(UTC).replace(tzinfo=None)
    return parsed

from_ms(value)

Convert epoch milliseconds, or seconds when the number is small, to naive UTC.

Source code in packages/agent-history/src/pdum/agent_history/schema.py

def from_ms(value: Any) -> datetime | None:
    """Convert epoch milliseconds, or seconds when the number is small, to naive UTC."""
    if value is None:
        return None
    try:
        number = float(value)
    except (TypeError, ValueError):
        return None
    if number <= 0:
        return None
    if number < 1e11:
        number *= 1000
    return datetime.fromtimestamp(number / 1000, tz=UTC).replace(tzinfo=None)

insert(connection, table, rows)

Insert rows as one Arrow batch when pyarrow is present, else statement by statement.

Parameters:

Name	Type	Description	Default
connection	DuckDBPyConnection	Open connection.	required
table	str	Target table.	required
rows	list of tuple	Rows built with :func:row.	required

Returns:

Type	Description
int	Number of rows inserted.

Source code in packages/agent-history/src/pdum/agent_history/schema.py

def insert(connection: Any, table: str, rows: list[tuple]) -> int:
    """Insert rows as one Arrow batch when pyarrow is present, else statement by statement.

    Parameters
    ----------
    connection : duckdb.DuckDBPyConnection
        Open connection.
    table : str
        Target table.
    rows : list of tuple
        Rows built with :func:`row`.

    Returns
    -------
    int
        Number of rows inserted.
    """
    if not rows:
        return 0
    if _pa is not None:
        names = [name for name, _ in COLUMNS[table]]
        arrow_schema = _pa.schema([(name, _ARROW_TYPES[kind]()) for name, kind in COLUMNS[table]])
        columns = list(zip(*rows, strict=True))
        batch = _pa.table(
            {
                name: _pa.array(values, type=arrow_schema.field(name).type)
                for name, values in zip(names, columns, strict=True)
            }
        )
        connection.register("_insert_batch", batch)
        connection.execute(f"INSERT INTO {table} SELECT * FROM _insert_batch")
        connection.unregister("_insert_batch")
        return len(rows)
    placeholders = ", ".join("?" for _ in COLUMNS[table])
    connection.executemany(f"INSERT INTO {table} VALUES ({placeholders})", rows)
    return len(rows)

row(table, **values)

Return a tuple in column order; unnamed columns are NULL.

Parameters:

Name	Type	Description	Default
table	str	Table name in :data:COLUMNS.	required
**values	Any	Column values by name.	{}

Source code in packages/agent-history/src/pdum/agent_history/schema.py

def row(table: str, **values: Any) -> tuple:
    """Return a tuple in column order; unnamed columns are NULL.

    Parameters
    ----------
    table : str
        Table name in :data:`COLUMNS`.
    **values : Any
        Column values by name.
    """
    names = [name for name, _ in COLUMNS[table]]
    unknown = set(values) - set(names)
    if unknown:
        raise KeyError(f"{table} has no columns {sorted(unknown)}")
    return tuple(values.get(name) for name in names)

Read a Codex home into the shared DuckDB schema without writing to the home.

Inputs are the thread catalogue (state_5.sqlite), the history projection (thread_history_1.sqlite), the log database (logs_2.sqlite) and every rollout file under sessions/ and archived_sessions/. Rollouts are the record: they alone hold per-response usage, rate limits, turn settings and fork lineage. Threads older than the support window are listed in coverage with in_window = false and otherwise skipped.

A unit of incremental work is one thread: its rollout file plus its catalogue row, whose updated_at and archived flag are the change fingerprint. The catalogue, projection and logs are small databases that every run reads, but rows are written only for the units being replaced; logs are appended from the target's last timestamp.

DEFAULT_WINDOW_DAYS = 183 module-attribute

Roughly six months; older history is counted, not modelled.

ROLLOUT_NAME = re.compile('^rollout-\\d{4}-\\d{2}-\\d{2}T\\d{2}-\\d{2}-\\d{2}-(.+)$') module-attribute

rollout-<timestamp>-<thread id>; the id after the timestamp names the thread.

CodexIngest

Bases: BaseIngest

One ingest of a Codex home into a target.

Parameters:

Name	Type	Description	Default
home	Path	Codex home to read.	required
target	Target or Path or str	Local DuckDB file or md:<database> to publish into.	required
window_days	int or None	Threads created before now - window_days are counted, not modelled. None disables.	DEFAULT_WINDOW_DAYS
blobs	bool	Store full item JSON in item_blobs.	True
now	datetime or None	Reference time for the window; defaults to the current UTC time.	None
mode	str	rebuild, sync or full; see :class:~pdum.agent_history.base.BaseIngest.	'rebuild'

Source code in packages/agent-history/src/pdum/agent_history/codex.py

class CodexIngest(BaseIngest):
    """One ingest of a Codex home into a target.

    Parameters
    ----------
    home : Path
        Codex home to read.
    target : Target or Path or str
        Local DuckDB file or ``md:<database>`` to publish into.
    window_days : int or None
        Threads created before ``now - window_days`` are counted, not modelled. None disables.
    blobs : bool
        Store full item JSON in ``item_blobs``.
    now : datetime or None
        Reference time for the window; defaults to the current UTC time.
    mode : str
        ``rebuild``, ``sync`` or ``full``; see :class:`~pdum.agent_history.base.BaseIngest`.
    """

    def __init__(
        self,
        home: Path,
        target: Target | Path | str,
        *,
        window_days: int | None = DEFAULT_WINDOW_DAYS,
        blobs: bool = True,
        now: datetime | None = None,
        mode: str = "rebuild",
        host: dict[str, str] | None = None,
    ) -> None:
        super().__init__(
            SOURCE, "codex", home, target, window_days=window_days, blobs=blobs, now=now, host=host, mode=mode
        )
        self.home = self.root
        self.threads: dict[str, dict[str, Any]] = {}
        self.turns: dict[tuple[str, str], dict[str, Any]] = {}
        self.units: dict[str, Unit] = {}
        self.window_by_thread: dict[str, bool] = {}
        self.logs_watermark: datetime | None = None
        self.rollout_items: dict[str, list[tuple[str | None, dict[str, Any], datetime | None, int]]] = {}
        """``item_completed`` payloads per thread, the fallback when the projection never materialized."""

    # ------------------------------------------------------------------- scan
    def scan(self) -> dict[str, Unit]:
        """List every thread as a unit: its rollout file, if any, plus its catalogue fingerprint."""
        rollouts: dict[str, Path] = {}
        for path in sorted((self.home / "sessions").glob("*/*/*/rollout-*.jsonl")) + sorted(
            (self.home / "archived_sessions").glob("rollout-*.jsonl")
        ):
            match = ROLLOUT_NAME.match(path.stem)
            rollouts[match.group(1) if match else path.stem] = path
        fingerprints: dict[str, str] = {}
        state = self.home / "state_5.sqlite"
        if state.exists():
            connection, _ = open_readonly(state)
            try:
                for thread_id, updated_ms, updated, archived in connection.execute(
                    "select id, updated_at_ms, updated_at, archived from threads"
                ):
                    fingerprints[thread_id] = f"{updated_ms or updated}|{int(bool(archived))}"
            finally:
                connection.close()
        units: dict[str, Unit] = {}
        for thread_id in sorted(set(rollouts) | set(fingerprints)):
            path = rollouts.get(thread_id)
            fingerprint = fingerprints.get(thread_id)
            if path is not None:
                units[thread_id] = Unit.from_path(thread_id, path, fingerprint)
            else:
                units[thread_id] = Unit(thread_id, None, fingerprint=fingerprint)
        return units

    def prepare(self, target: duckdb.DuckDBPyConnection) -> None:
        """Remember the newest log already published, so a sync appends only newer log rows."""
        row = target.execute("select max(ts) from logs where source_id = ?", [self.source_id]).fetchone()
        self.logs_watermark = row[0] if row else None

    # ----------------------------------------------------------------- ingest
    def ingest(self, units: dict[str, Unit]) -> None:
        """Read the catalogue, the given units' rollouts, their projection rows and new logs."""
        self.units = units
        self.catalogue()
        self.rollouts()
        self.projection()
        self.logs()
        self.finish()

    # -------------------------------------------------------------- catalogue
    def catalogue(self) -> None:
        assert self.con is not None
        started = time.perf_counter()
        path = self.home / "state_5.sqlite"
        if not path.exists():
            self.summary.notes.append("state_5.sqlite absent")
            self.phase("catalogue", started, 0, "absent")
            return
        connection, how = open_readonly(path)
        connection.row_factory = sqlite3.Row
        count = 0
        for record in connection.execute("select * from threads"):
            created = schema.from_ms(record["created_at_ms"] or record["created_at"])
            in_window = self.in_window(created)
            self.window_by_thread[record["id"]] = in_window
            if record["id"] not in self.units:
                continue
            source = record["source"] or ""
            source_kind = source
            if source.startswith("{"):
                try:
                    parsed = json.loads(source)
                    source_kind = next(iter(parsed)) if isinstance(parsed, dict) and parsed else source[:32]
                except ValueError:
                    source_kind = source[:32]
            self.threads[record["id"]] = {
                "thread_id": record["id"],
                "created_at": created,
                "updated_at": schema.from_ms(record["updated_at_ms"] or record["updated_at"]),
                "source_kind": source_kind,
                "thread_source": record["thread_source"],
                "originator": record["originator"],
                "cli_version": record["cli_version"],
                "model_provider": record["model_provider"],
                "repo": repo_from_origin(record["git_origin_url"]),
                "cwd": record["cwd"],
                "git_branch": record["git_branch"],
                "git_origin": record["git_origin_url"],
                "git_sha": record["git_sha"],
                "title": record["title"],
                "preview_len": len(record["preview"] or ""),
                "model": record["model"],
                "reasoning_effort": record["reasoning_effort"],
                "archived": bool(record["archived"]),
                "history_mode": record["history_mode"],
                "tokens_used": record["tokens_used"],
                "agent_nickname": record["agent_nickname"],
                "agent_role": record["agent_role"],
                "agent_path": record["agent_path"],
                "rollout_path": record["rollout_path"],
                "in_window": in_window,
                "origin": "state_5",
            }
            cover = self.cover(record["id"])
            cover["in_catalogue"] = True
            cover["in_window"] = in_window
            self.unit_threads.setdefault(record["id"], [record["id"]])
            if record["cli_version"]:
                self.versions.add(str(record["cli_version"]))
            count += 1
        edges = [
            schema.row(
                "spawn_edges",
                parent_thread_id=parent,
                child_thread_id=child,
                status=status,
                source=SOURCE,
                origin="state_5",
            )
            for parent, child, status in connection.execute(
                "select parent_thread_id, child_thread_id, status from thread_spawn_edges"
            )
            if child in self.units
        ]
        self.write("spawn_edges", edges)
        connection.close()
        self.phase("catalogue", started, count, f"{how}; spawn edges {len(edges)}")

    # --------------------------------------------------------------- rollouts
    def rollouts(self) -> None:
        assert self.con is not None
        started = time.perf_counter()
        files = [(unit_id, unit.path) for unit_id, unit in sorted(self.units.items()) if unit.path is not None]
        compressed = len(list((self.home / "sessions").glob("*/*/*/rollout-*.zst")))
        compressed += len(list((self.home / "archived_sessions").glob("rollout-*.zst")))
        total_bytes = 0
        total_lines = 0
        skipped = 0
        for unit_id, path in files:
            total_bytes += path.stat().st_size
            lines = self.rollout(path, unit_id)
            if lines is None:
                skipped += 1
            else:
                total_lines += lines
        self.phase(
            "rollouts",
            started,
            total_lines,
            f"files {len(files)}; bytes {total_bytes}; outside window {skipped}; compressed skipped {compressed}",
        )

    def rollout(self, path: Path, unit_id: str) -> int | None:
        """Read one rollout; return its line count, or None when outside the window."""
        assert self.con is not None
        responses: list[tuple] = []
        counts: list[tuple] = []
        events: list[tuple] = []
        response_items: list[tuple] = []
        thread_id: str | None = None
        current_turn: str | None = None
        lines = 0
        fmt = "envelope"
        cover: dict[str, Any] | None = None
        with path.open("rb") as stream:
            for number, raw in enumerate(stream):
                lines += 1
                try:
                    record = json.loads(raw)
                except ValueError:
                    if cover is not None:
                        cover["unknown_line_types"] += 1
                        self.unknown("unparseable")
                    continue
                if number == 0:
                    if record.get("type") == "session_meta":
                        payload = record.get("payload") or {}
                        thread_id = payload.get("id")
                    else:
                        fmt = "legacy"
                        thread_id = record.get("id") or unit_id
                        payload = record
                    if not thread_id:
                        return lines
                    if thread_id != unit_id:
                        self.summary.notes.append(f"rollout {path.name} names thread {thread_id}")
                    self.unit_threads[unit_id] = sorted({unit_id, thread_id})
                    cover = self.cover(thread_id)
                    cover["rollout_format"] = fmt
                    anchor = self.session_meta(thread_id, payload, path, fmt)
                    in_window = self.in_window(anchor)
                    cover["in_window"] = in_window
                    self.threads[thread_id]["in_window"] = in_window
                    if not in_window:
                        cover["rollout_lines"] = None
                        return None
                    if fmt == "legacy":
                        continue
                if cover is None or thread_id is None:
                    continue
                if fmt == "legacy":
                    kind = record.get("type") or record.get("record_type")
                    if kind in LEGACY_ITEM_TYPES:
                        response_items.append(
                            schema.row(
                                "response_items",
                                thread_id=thread_id,
                                item_type=kind,
                                role=record.get("role"),
                                name=record.get("name"),
                                call_id=record.get("call_id"),
                                text_len=self.content_length(record),
                                source=SOURCE,
                                origin="rollout-legacy",
                            )
                        )
                        cover["response_items"] += 1
                    else:
                        cover["unknown_line_types"] += 1
                        self.unknown(f"legacy:{kind}")
                    continue
                kind = record.get("type")
                payload = record.get("payload") or {}
                ts = schema.from_iso(record.get("timestamp"))
                ordinal = record.get("ordinal")
                if kind == "turn_context":
                    cover["turn_contexts"] += 1
                    current_turn = payload.get("turn_id") or current_turn
                    self.turn_context(thread_id, payload, ordinal)
                elif kind == "token_usage_record":
                    cover["usage_records"] += 1
                    turn_key = (thread_id, payload.get("turn_id") or "")
                    responses.append(
                        schema.row(
                            "responses",
                            thread_id=thread_id,
                            turn_id=payload.get("turn_id"),
                            root_turn_id=payload.get("root_turn_id"),
                            session_id=payload.get("session_id"),
                            response_id=payload.get("response_id"),
                            ts=ts,
                            rollout_ordinal=ordinal,
                            model=self.turns.get(turn_key, {}).get("model"),
                            **_usage_fields("", payload.get("usage")),
                            **_usage_fields(
                                "turn_",
                                payload.get("turn_token_usage"),
                                ("input_tokens", "cached_input_tokens", "output_tokens", "total_tokens"),
                            ),
                            **_usage_fields(
                                "thread_",
                                payload.get("thread_token_usage"),
                                ("input_tokens", "cached_input_tokens", "output_tokens", "total_tokens"),
                            ),
                            source=SOURCE,
                            origin="rollout",
                        )
                    )
                elif kind == "event_msg":
                    sub = payload.get("type")
                    if sub == "token_count":
                        cover["token_counts"] += 1
                        counts.append(self.token_count(thread_id, current_turn, payload, ts, ordinal))
                    elif sub == "item_completed":
                        cover["rollout_item_completed"] += 1
                        self.rollout_items.setdefault(thread_id, []).append(
                            (payload.get("turn_id") or current_turn, payload.get("item") or {}, ts, ordinal)
                        )
                    else:
                        if sub in {"task_started", "task_complete", "turn_aborted"}:
                            current_turn = payload.get("turn_id") or current_turn
                        self.event(thread_id, payload, ts, ordinal, sub, events)
                elif kind == "response_item":
                    cover["response_items"] += 1
                    meta = payload.get("internal_chat_message_metadata_passthrough") or {}
                    response_items.append(
                        schema.row(
                            "response_items",
                            thread_id=thread_id,
                            turn_id=meta.get("turn_id") or current_turn,
                            ts=ts,
                            create_time=schema.from_ms(meta.get("create_time")),
                            rollout_ordinal=ordinal,
                            item_type=payload.get("type"),
                            role=payload.get("role"),
                            name=payload.get("name"),
                            call_id=payload.get("call_id"),
                            text_len=self.content_length(payload),
                            source=SOURCE,
                            origin="rollout",
                        )
                    )
                elif kind == "compacted":
                    cover["compactions"] += 1
                    history = payload.get("replacement_history") or []
                    events.append(
                        schema.row(
                            "rollout_events",
                            thread_id=thread_id,
                            turn_id=current_turn,
                            ts=ts,
                            rollout_ordinal=ordinal,
                            event_type="compacted",
                            detail=json.dumps({"replacement_history": len(history)}),
                            source=SOURCE,
                            origin="rollout",
                        )
                    )
                elif kind in PASSTHROUGH_EVENTS:
                    events.append(
                        schema.row(
                            "rollout_events",
                            thread_id=thread_id,
                            turn_id=current_turn,
                            ts=ts,
                            rollout_ordinal=ordinal,
                            event_type=kind,
                            source=SOURCE,
                            origin="rollout",
                        )
                    )
                else:
                    cover["unknown_line_types"] += 1
                    self.unknown(str(kind))
        if cover is not None:
            cover["rollout_lines"] = lines
        self.write("responses", responses)
        self.write("token_counts", counts)
        self.write("rollout_events", events)
        self.write("response_items", response_items)
        return lines

    @staticmethod
    def content_length(record: dict[str, Any]) -> int:
        total = 0
        for key in ("content", "output", "input", "arguments", "summary"):
            value = record.get(key)
            if isinstance(value, str):
                total += len(value)
            elif isinstance(value, list):
                for part in value:
                    if isinstance(part, dict) and isinstance(part.get("text"), str):
                        total += len(part["text"])
                    elif isinstance(part, str):
                        total += len(part)
        return total

    def session_meta(self, thread_id: str, payload: dict[str, Any], path: Path, fmt: str) -> datetime | None:
        """Merge a rollout's first line into the thread record; return the window anchor."""
        git = payload.get("git") or {}
        thread = self.threads.get(thread_id)
        if thread is None:
            thread = self.threads[thread_id] = {
                "thread_id": thread_id,
                "origin": "rollout-legacy" if fmt == "legacy" else "rollout",
                "archived": "archived_sessions" in path.parts,
            }
        thread.update(
            session_id=payload.get("session_id"),
            forked_from_id=payload.get("forked_from_id"),
            parent_thread_id=payload.get("parent_thread_id"),
            rollout_bytes=path.stat().st_size,
            rollout_format=fmt,
        )
        meta_time = schema.from_iso(payload.get("timestamp"))
        thread.setdefault("created_at", meta_time)
        thread.setdefault("cwd", payload.get("cwd"))
        thread.setdefault("originator", payload.get("originator"))
        thread.setdefault("cli_version", payload.get("cli_version"))
        thread.setdefault("model_provider", payload.get("model_provider"))
        thread.setdefault("history_mode", payload.get("history_mode"))
        for key, value in (
            ("git_branch", git.get("branch")),
            ("git_origin", git.get("repository_url")),
            ("git_sha", git.get("commit_hash")),
        ):
            if not thread.get(key) and value:
                thread[key] = value
        if not thread.get("repo"):
            thread["repo"] = repo_from_origin(thread.get("git_origin"))
        if payload.get("cli_version"):
            self.versions.add(str(payload["cli_version"]))
        thread.setdefault("rollout_path", str(path))
        if thread.get("source_kind") is None:
            source = payload.get("source")
            thread["source_kind"] = next(iter(source)) if isinstance(source, dict) and source else source
        thread.setdefault("thread_source", payload.get("thread_source"))
        thread.setdefault("agent_nickname", payload.get("agent_nickname"))
        thread.setdefault("agent_role", payload.get("agent_role"))
        anchor = thread.get("created_at") or meta_time
        if anchor is None:
            anchor = datetime.fromtimestamp(path.stat().st_mtime, tz=UTC).replace(tzinfo=None)
        return anchor

    def turn_context(self, thread_id: str, payload: dict[str, Any], ordinal: int | None) -> None:
        turn_id = payload.get("turn_id")
        if not turn_id:
            return
        turn = self.turns.setdefault(
            (thread_id, turn_id), {"thread_id": thread_id, "turn_id": turn_id, "origin": "rollout"}
        )
        if turn.get("model"):
            return
        settings = (payload.get("collaboration_mode") or {}).get("settings") or {}
        sandbox = payload.get("sandbox_policy") or {}
        turn.update(
            root_turn_id=turn.get("root_turn_id") or payload.get("root_turn_id"),
            rollout_ordinal=turn.get("rollout_ordinal") if turn.get("rollout_ordinal") is not None else ordinal,
            model=payload.get("model") or settings.get("model"),
            reasoning_effort=payload.get("effort") or settings.get("reasoning_effort"),
            cwd=payload.get("cwd"),
            approval_policy=payload.get("approval_policy"),
            sandbox_type=sandbox.get("type") if isinstance(sandbox, dict) else str(sandbox),
            personality=payload.get("personality"),
            collaboration_mode=(payload.get("collaboration_mode") or {}).get("mode"),
        )

    def token_count(
        self, thread_id: str, turn_id: str | None, payload: dict[str, Any], ts: Any, ordinal: int | None
    ) -> tuple:
        info = payload.get("info") or {}
        limits = payload.get("rate_limits") or {}
        primary = limits.get("primary") or {}
        secondary = limits.get("secondary") or {}
        credits = limits.get("credits") or {}
        balance = credits.get("balance")
        try:
            balance = float(balance) if balance is not None else None
        except (TypeError, ValueError):
            balance = None
        totals = _usage_fields("total_", info.get("total_token_usage"))
        totals["total_tokens"] = totals.pop("total_total_tokens")
        return schema.row(
            "token_counts",
            thread_id=thread_id,
            turn_id=turn_id,
            ts=ts,
            rollout_ordinal=ordinal,
            **totals,
            **_usage_fields("last_", info.get("last_token_usage")),
            model_context_window=info.get("model_context_window"),
            limit_id=limits.get("limit_id"),
            plan_type=limits.get("plan_type"),
            primary_used_percent=primary.get("used_percent"),
            primary_window_minutes=primary.get("window_minutes"),
            primary_resets_at=schema.from_ms(primary.get("resets_at")),
            secondary_used_percent=secondary.get("used_percent"),
            secondary_window_minutes=secondary.get("window_minutes"),
            secondary_resets_at=schema.from_ms(secondary.get("resets_at")),
            credits_balance=balance,
            credits_unlimited=credits.get("unlimited"),
            rate_limit_reached_type=limits.get("rate_limit_reached_type"),
            source=SOURCE,
            origin="rollout",
        )

    def event(
        self,
        thread_id: str,
        payload: dict[str, Any],
        ts: Any,
        ordinal: int | None,
        sub: str | None,
        events: list[tuple],
    ) -> None:
        turn_id = payload.get("turn_id")
        detail: dict[str, Any] = {}
        turn = None
        if turn_id and sub in {"task_started", "task_complete", "turn_aborted"}:
            turn = self.turns.setdefault(
                (thread_id, turn_id), {"thread_id": thread_id, "turn_id": turn_id, "origin": "rollout"}
            )
        if sub == "task_started":
            detail = {
                "model_context_window": payload.get("model_context_window"),
                "collaboration_mode_kind": payload.get("collaboration_mode_kind"),
            }
            if turn is not None:
                turn.setdefault("root_turn_id", payload.get("root_turn_id"))
                turn["model_context_window"] = payload.get("model_context_window")
                if not turn.get("started_at"):
                    turn["started_at"] = schema.from_ms(payload.get("started_at")) or ts
        elif sub == "task_complete":
            detail = {
                "duration_ms": payload.get("duration_ms"),
                "time_to_first_token_ms": payload.get("time_to_first_token_ms"),
                "last_agent_message_len": len(payload.get("last_agent_message") or ""),
            }
            if turn is not None:
                turn["time_to_first_token_ms"] = payload.get("time_to_first_token_ms")
                if not turn.get("completed_at"):
                    turn["completed_at"] = schema.from_ms(payload.get("completed_at")) or ts
                if turn.get("duration_ms") is None:
                    turn["duration_ms"] = payload.get("duration_ms")
                turn.setdefault("status", "completed")
        elif sub == "turn_aborted":
            detail = {"reason": payload.get("reason")}
            if turn is not None:
                turn.setdefault("status", "interrupted")
        elif sub == "thread_settings_applied":
            settings = payload.get("thread_settings") or {}
            detail = {
                "model": settings.get("model"),
                "reasoning_effort": settings.get("reasoning_effort"),
                "service_tier": settings.get("service_tier"),
                "personality": settings.get("personality"),
            }
        else:
            detail = {key: payload.get(key) for key in ("reason", "message", "kind") if key in payload}
        events.append(
            schema.row(
                "rollout_events",
                thread_id=thread_id,
                turn_id=turn_id,
                ts=ts,
                rollout_ordinal=ordinal,
                event_type=sub,
                detail=json.dumps(detail) if detail else None,
                source=SOURCE,
                origin="rollout",
            )
        )

    # ------------------------------------------------------------- projection
    def projection(self) -> None:
        assert self.con is not None
        started = time.perf_counter()
        path = self.home / "thread_history_1.sqlite"
        if not path.exists():
            self.summary.notes.append("thread_history_1.sqlite absent")
            derived = self.items_from_rollouts()
            self.phase("projection", started, derived, f"absent; items from rollout events {derived}")
            return
        connection, how = open_readonly(path)
        connection.row_factory = sqlite3.Row
        tables = {row[0] for row in connection.execute("select name from sqlite_master where type = 'table'")}
        if "thread_turns" not in tables or "thread_items" not in tables:
            # A home that only ever ran `codex exec` has the file but never materialized the projection.
            connection.close()
            self.summary.notes.append("thread_history_1.sqlite has no projection tables")
            derived = self.items_from_rollouts()
            self.phase("projection", started, derived, f"{how}; unmaterialized; items from rollout events {derived}")
            return
        turn_rows = 0
        skipped_turns = 0
        for record in connection.execute("select * from thread_turns"):
            if not self.thread_in_window(record["thread_id"], record["started_at"]):
                skipped_turns += 1
                continue
            error_message = error_kind = None
            if record["error_json"]:
                try:
                    error = json.loads(record["error_json"])
                    error_message = (error.get("message") or "")[:500]
                    info = error.get("codexErrorInfo")
                    error_kind = info if isinstance(info, str) else json.dumps(info)[:120] if info else None
                except ValueError:
                    error_message = record["error_json"][:500]
            key = (record["thread_id"], record["turn_id"])
            existing = self.turns.get(key, {})
            existing.update(
                {
                    "thread_id": record["thread_id"],
                    "turn_id": record["turn_id"],
                    "root_turn_id": existing.get("root_turn_id") or record["root_turn_id"],
                    "rollout_ordinal": record["rollout_ordinal"],
                    "status": record["status"],
                    "error_message": error_message,
                    "error_kind": error_kind,
                    "started_at": schema.from_ms(record["started_at"]) or existing.get("started_at"),
                    "completed_at": schema.from_ms(record["completed_at"]) or existing.get("completed_at"),
                    "duration_ms": record["duration_ms"]
                    if record["duration_ms"] is not None
                    else existing.get("duration_ms"),
                    "origin": "thread_history_1+rollout" if existing.get("model") else "thread_history_1",
                }
            )
            self.turns[key] = existing
            cover = self.cover(record["thread_id"])
            cover["in_projection"] = True
            cover["projection_turns"] += 1
            turn_rows += 1
        batch: list[tuple] = []
        blobs: list[tuple] = []
        item_rows = 0
        skipped_items = 0
        for chunk in _chunks(sorted(self.units)):
            placeholders = ", ".join("?" for _ in chunk)
            cursor = connection.execute(
                "select thread_id, turn_id, item_id, rollout_ordinal, created_at_ms, started_at_ms, completed_at_ms, "
                f"item_type, item_json from thread_items where thread_id in ({placeholders})",
                chunk,
            )
            while True:
                rows = cursor.fetchmany(2000)
                if not rows:
                    break
                for record in rows:
                    if not self.thread_in_window(record["thread_id"], record["created_at_ms"]):
                        skipped_items += 1
                        continue
                    raw = record["item_json"]
                    try:
                        item = json.loads(raw)
                    except ValueError:
                        item = {"type": record["item_type"], "id": record["item_id"]}
                    columns = item_columns.extract(item)
                    duration = None
                    if record["started_at_ms"] and record["completed_at_ms"]:
                        duration = record["completed_at_ms"] - record["started_at_ms"]
                    columns.update(
                        thread_id=record["thread_id"],
                        turn_id=record["turn_id"],
                        item_id=record["item_id"],
                        rollout_ordinal=record["rollout_ordinal"],
                        item_type=record["item_type"] or columns.get("item_type"),
                        created_at=schema.from_ms(record["created_at_ms"]),
                        started_at=schema.from_ms(record["started_at_ms"]),
                        completed_at=schema.from_ms(record["completed_at_ms"]),
                        duration_ms=duration,
                        item_json_bytes=len(raw),
                        source=SOURCE,
                        origin="thread_history_1",
                    )
                    batch.append(schema.row("items", **columns))
                    if self.blobs:
                        blobs.append(
                            schema.row(
                                "item_blobs",
                                thread_id=record["thread_id"],
                                turn_id=record["turn_id"],
                                item_id=record["item_id"],
                                item_json=raw,
                                source=SOURCE,
                                origin="thread_history_1",
                            )
                        )
                    cover = self.cover(record["thread_id"])
                    cover["in_projection"] = True
                    cover["projection_items"] += 1
                    item_rows += 1
                self.write("items", batch)
                self.write("item_blobs", blobs)
                batch.clear()
                blobs.clear()
        connection.close()
        derived = self.items_from_rollouts()
        self.phase(
            "projection",
            started,
            item_rows + derived,
            f"{how}; turns {turn_rows}; blobs {self.blobs}; outside window turns {skipped_turns} "
            f"items {skipped_items}; items from rollout events {derived}",
        )

    def items_from_rollouts(self) -> int:
        """Write items from rollout ``item_completed`` events for threads the projection does not cover.

        A home that only ran ``codex exec`` has rollouts but no materialized projection, so its tool
        calls would otherwise be lost. The core's item shape is normalized to the v2 shape first.
        Returns the number of items written.
        """
        rows: list[tuple] = []
        blobs: list[tuple] = []
        threads = 0
        for thread_id, entries in self.rollout_items.items():
            cover = self.cover(thread_id)
            if cover["projection_items"] > 0 or not cover["in_window"]:
                continue
            threads += 1
            for turn_id, raw_item, ts, ordinal in entries:
                item = item_columns.normalize_core_item(raw_item)
                columns = item_columns.extract(item)
                duration = columns.get("exec_duration_ms")
                started_at = ts - timedelta(milliseconds=duration) if ts is not None and duration else None
                item_id = item.get("id") or f"{thread_id}:{ordinal}"
                columns.update(
                    thread_id=thread_id,
                    turn_id=turn_id,
                    item_id=item_id,
                    rollout_ordinal=ordinal,
                    created_at=ts,
                    started_at=started_at,
                    completed_at=ts,
                    duration_ms=duration,
                    item_json_bytes=len(json.dumps(raw_item)),
                    source=SOURCE,
                    origin="rollout",
                )
                rows.append(schema.row("items", **columns))
                if self.blobs:
                    blobs.append(
                        schema.row(
                            "item_blobs",
                            thread_id=thread_id,
                            turn_id=turn_id,
                            item_id=item_id,
                            item_json=json.dumps(item),
                            source=SOURCE,
                            origin="rollout",
                        )
                    )
        self.write("items", rows)
        self.write("item_blobs", blobs)
        if rows:
            self.summary.notes.append(f"items of {threads} threads come from rollout item_completed events")
        return len(rows)

    def thread_in_window(self, thread_id: str, fallback_ms: Any) -> bool:
        """Decide membership for a projection row: only this run's units, by their window, else by timestamp."""
        if thread_id not in self.units:
            return False
        cover = self.coverage.get(thread_id)
        if cover is not None:
            return bool(cover["in_window"])
        if thread_id in self.window_by_thread:
            return self.window_by_thread[thread_id]
        return self.in_window(schema.from_ms(fallback_ms))

    # ------------------------------------------------------------------- logs
    def logs(self) -> None:
        assert self.con is not None
        started = time.perf_counter()
        path = self.home / "logs_2.sqlite"
        if not path.exists():
            self.summary.notes.append("logs_2.sqlite absent")
            self.phase("logs", started, 0, "absent")
            return
        connection, how = open_readonly(path)
        rows = []
        if self.logs_watermark is not None:
            since = self.logs_watermark.replace(tzinfo=UTC).timestamp()
            query = (
                "select ts, ts_nanos, level, target, feedback_log_body, thread_id, process_uuid from logs where ts > ?"
            )
        else:
            since = self.cutoff.replace(tzinfo=UTC).timestamp() if self.cutoff else 0
            query = (
                "select ts, ts_nanos, level, target, feedback_log_body, thread_id, process_uuid from logs where ts >= ?"
            )
        for ts, nanos, level, target, body, thread_id, process_uuid in connection.execute(query, (since,)):
            moment = datetime.fromtimestamp(ts + (nanos or 0) / 1e9, tz=UTC).replace(tzinfo=None) if ts else None
            rows.append(
                schema.row(
                    "logs",
                    ts=moment,
                    level=level,
                    target=target,
                    thread_id=thread_id,
                    process_uuid=process_uuid,
                    body=(body or "")[:2000] if level in {"WARN", "ERROR"} else None,
                    source=SOURCE,
                    origin="logs_2",
                )
            )
        connection.close()
        self.write("logs", rows)
        self.phase("logs", started, len(rows), f"{how}; since watermark {self.logs_watermark is not None}")

    # ----------------------------------------------------------------- finish
    def finish(self) -> None:
        assert self.con is not None
        started = time.perf_counter()
        thread_columns = dict(schema.COLUMNS["threads"])
        thread_rows = []
        links = []
        for thread in self.threads.values():
            thread.setdefault("source", SOURCE)
            thread.setdefault("in_window", True)
            thread_rows.append(schema.row("threads", **{k: v for k, v in thread.items() if k in thread_columns}))
            parent = thread.get("forked_from_id")
            if parent and parent != thread["thread_id"]:
                # The rollout's session_meta names the fork parent; the catalogue does not record forks.
                links.append(
                    schema.row(
                        "thread_links",
                        thread_id=thread["thread_id"],
                        kind="forked-from",
                        target_thread_id=parent,
                        ordinal=0,
                        ts=thread.get("created_at"),
                        source=SOURCE,
                        origin=thread.get("origin"),
                    )
                )
        self.write("threads", thread_rows)
        self.write("thread_links", links)
        turn_rows = [
            schema.row("turns", **{**turn, "source": SOURCE})
            for turn in self.turns.values()
            if self.coverage.get(turn["thread_id"], {}).get("in_window", True)
        ]
        self.write("turns", turn_rows)
        cover_rows = self.write_coverage_and_meta()
        self.phase(
            "finish",
            started,
            len(thread_rows) + len(turn_rows) + cover_rows,
            f"threads {len(thread_rows)}; turns {len(turn_rows)}; coverage {cover_rows}",
        )

rollout_items = {} instance-attribute

item_completed payloads per thread, the fallback when the projection never materialized.

ingest(units)

Read the catalogue, the given units' rollouts, their projection rows and new logs.

Source code in packages/agent-history/src/pdum/agent_history/codex.py

def ingest(self, units: dict[str, Unit]) -> None:
    """Read the catalogue, the given units' rollouts, their projection rows and new logs."""
    self.units = units
    self.catalogue()
    self.rollouts()
    self.projection()
    self.logs()
    self.finish()

items_from_rollouts()

Write items from rollout item_completed events for threads the projection does not cover.

A home that only ran codex exec has rollouts but no materialized projection, so its tool calls would otherwise be lost. The core's item shape is normalized to the v2 shape first. Returns the number of items written.

Source code in packages/agent-history/src/pdum/agent_history/codex.py

def items_from_rollouts(self) -> int:
    """Write items from rollout ``item_completed`` events for threads the projection does not cover.

    A home that only ran ``codex exec`` has rollouts but no materialized projection, so its tool
    calls would otherwise be lost. The core's item shape is normalized to the v2 shape first.
    Returns the number of items written.
    """
    rows: list[tuple] = []
    blobs: list[tuple] = []
    threads = 0
    for thread_id, entries in self.rollout_items.items():
        cover = self.cover(thread_id)
        if cover["projection_items"] > 0 or not cover["in_window"]:
            continue
        threads += 1
        for turn_id, raw_item, ts, ordinal in entries:
            item = item_columns.normalize_core_item(raw_item)
            columns = item_columns.extract(item)
            duration = columns.get("exec_duration_ms")
            started_at = ts - timedelta(milliseconds=duration) if ts is not None and duration else None
            item_id = item.get("id") or f"{thread_id}:{ordinal}"
            columns.update(
                thread_id=thread_id,
                turn_id=turn_id,
                item_id=item_id,
                rollout_ordinal=ordinal,
                created_at=ts,
                started_at=started_at,
                completed_at=ts,
                duration_ms=duration,
                item_json_bytes=len(json.dumps(raw_item)),
                source=SOURCE,
                origin="rollout",
            )
            rows.append(schema.row("items", **columns))
            if self.blobs:
                blobs.append(
                    schema.row(
                        "item_blobs",
                        thread_id=thread_id,
                        turn_id=turn_id,
                        item_id=item_id,
                        item_json=json.dumps(item),
                        source=SOURCE,
                        origin="rollout",
                    )
                )
    self.write("items", rows)
    self.write("item_blobs", blobs)
    if rows:
        self.summary.notes.append(f"items of {threads} threads come from rollout item_completed events")
    return len(rows)

prepare(target)

Remember the newest log already published, so a sync appends only newer log rows.

Source code in packages/agent-history/src/pdum/agent_history/codex.py

def prepare(self, target: duckdb.DuckDBPyConnection) -> None:
    """Remember the newest log already published, so a sync appends only newer log rows."""
    row = target.execute("select max(ts) from logs where source_id = ?", [self.source_id]).fetchone()
    self.logs_watermark = row[0] if row else None

rollout(path, unit_id)

Read one rollout; return its line count, or None when outside the window.

Source code in packages/agent-history/src/pdum/agent_history/codex.py

def rollout(self, path: Path, unit_id: str) -> int | None:
    """Read one rollout; return its line count, or None when outside the window."""
    assert self.con is not None
    responses: list[tuple] = []
    counts: list[tuple] = []
    events: list[tuple] = []
    response_items: list[tuple] = []
    thread_id: str | None = None
    current_turn: str | None = None
    lines = 0
    fmt = "envelope"
    cover: dict[str, Any] | None = None
    with path.open("rb") as stream:
        for number, raw in enumerate(stream):
            lines += 1
            try:
                record = json.loads(raw)
            except ValueError:
                if cover is not None:
                    cover["unknown_line_types"] += 1
                    self.unknown("unparseable")
                continue
            if number == 0:
                if record.get("type") == "session_meta":
                    payload = record.get("payload") or {}
                    thread_id = payload.get("id")
                else:
                    fmt = "legacy"
                    thread_id = record.get("id") or unit_id
                    payload = record
                if not thread_id:
                    return lines
                if thread_id != unit_id:
                    self.summary.notes.append(f"rollout {path.name} names thread {thread_id}")
                self.unit_threads[unit_id] = sorted({unit_id, thread_id})
                cover = self.cover(thread_id)
                cover["rollout_format"] = fmt
                anchor = self.session_meta(thread_id, payload, path, fmt)
                in_window = self.in_window(anchor)
                cover["in_window"] = in_window
                self.threads[thread_id]["in_window"] = in_window
                if not in_window:
                    cover["rollout_lines"] = None
                    return None
                if fmt == "legacy":
                    continue
            if cover is None or thread_id is None:
                continue
            if fmt == "legacy":
                kind = record.get("type") or record.get("record_type")
                if kind in LEGACY_ITEM_TYPES:
                    response_items.append(
                        schema.row(
                            "response_items",
                            thread_id=thread_id,
                            item_type=kind,
                            role=record.get("role"),
                            name=record.get("name"),
                            call_id=record.get("call_id"),
                            text_len=self.content_length(record),
                            source=SOURCE,
                            origin="rollout-legacy",
                        )
                    )
                    cover["response_items"] += 1
                else:
                    cover["unknown_line_types"] += 1
                    self.unknown(f"legacy:{kind}")
                continue
            kind = record.get("type")
            payload = record.get("payload") or {}
            ts = schema.from_iso(record.get("timestamp"))
            ordinal = record.get("ordinal")
            if kind == "turn_context":
                cover["turn_contexts"] += 1
                current_turn = payload.get("turn_id") or current_turn
                self.turn_context(thread_id, payload, ordinal)
            elif kind == "token_usage_record":
                cover["usage_records"] += 1
                turn_key = (thread_id, payload.get("turn_id") or "")
                responses.append(
                    schema.row(
                        "responses",
                        thread_id=thread_id,
                        turn_id=payload.get("turn_id"),
                        root_turn_id=payload.get("root_turn_id"),
                        session_id=payload.get("session_id"),
                        response_id=payload.get("response_id"),
                        ts=ts,
                        rollout_ordinal=ordinal,
                        model=self.turns.get(turn_key, {}).get("model"),
                        **_usage_fields("", payload.get("usage")),
                        **_usage_fields(
                            "turn_",
                            payload.get("turn_token_usage"),
                            ("input_tokens", "cached_input_tokens", "output_tokens", "total_tokens"),
                        ),
                        **_usage_fields(
                            "thread_",
                            payload.get("thread_token_usage"),
                            ("input_tokens", "cached_input_tokens", "output_tokens", "total_tokens"),
                        ),
                        source=SOURCE,
                        origin="rollout",
                    )
                )
            elif kind == "event_msg":
                sub = payload.get("type")
                if sub == "token_count":
                    cover["token_counts"] += 1
                    counts.append(self.token_count(thread_id, current_turn, payload, ts, ordinal))
                elif sub == "item_completed":
                    cover["rollout_item_completed"] += 1
                    self.rollout_items.setdefault(thread_id, []).append(
                        (payload.get("turn_id") or current_turn, payload.get("item") or {}, ts, ordinal)
                    )
                else:
                    if sub in {"task_started", "task_complete", "turn_aborted"}:
                        current_turn = payload.get("turn_id") or current_turn
                    self.event(thread_id, payload, ts, ordinal, sub, events)
            elif kind == "response_item":
                cover["response_items"] += 1
                meta = payload.get("internal_chat_message_metadata_passthrough") or {}
                response_items.append(
                    schema.row(
                        "response_items",
                        thread_id=thread_id,
                        turn_id=meta.get("turn_id") or current_turn,
                        ts=ts,
                        create_time=schema.from_ms(meta.get("create_time")),
                        rollout_ordinal=ordinal,
                        item_type=payload.get("type"),
                        role=payload.get("role"),
                        name=payload.get("name"),
                        call_id=payload.get("call_id"),
                        text_len=self.content_length(payload),
                        source=SOURCE,
                        origin="rollout",
                    )
                )
            elif kind == "compacted":
                cover["compactions"] += 1
                history = payload.get("replacement_history") or []
                events.append(
                    schema.row(
                        "rollout_events",
                        thread_id=thread_id,
                        turn_id=current_turn,
                        ts=ts,
                        rollout_ordinal=ordinal,
                        event_type="compacted",
                        detail=json.dumps({"replacement_history": len(history)}),
                        source=SOURCE,
                        origin="rollout",
                    )
                )
            elif kind in PASSTHROUGH_EVENTS:
                events.append(
                    schema.row(
                        "rollout_events",
                        thread_id=thread_id,
                        turn_id=current_turn,
                        ts=ts,
                        rollout_ordinal=ordinal,
                        event_type=kind,
                        source=SOURCE,
                        origin="rollout",
                    )
                )
            else:
                cover["unknown_line_types"] += 1
                self.unknown(str(kind))
    if cover is not None:
        cover["rollout_lines"] = lines
    self.write("responses", responses)
    self.write("token_counts", counts)
    self.write("rollout_events", events)
    self.write("response_items", response_items)
    return lines

scan()

List every thread as a unit: its rollout file, if any, plus its catalogue fingerprint.

Source code in packages/agent-history/src/pdum/agent_history/codex.py

def scan(self) -> dict[str, Unit]:
    """List every thread as a unit: its rollout file, if any, plus its catalogue fingerprint."""
    rollouts: dict[str, Path] = {}
    for path in sorted((self.home / "sessions").glob("*/*/*/rollout-*.jsonl")) + sorted(
        (self.home / "archived_sessions").glob("rollout-*.jsonl")
    ):
        match = ROLLOUT_NAME.match(path.stem)
        rollouts[match.group(1) if match else path.stem] = path
    fingerprints: dict[str, str] = {}
    state = self.home / "state_5.sqlite"
    if state.exists():
        connection, _ = open_readonly(state)
        try:
            for thread_id, updated_ms, updated, archived in connection.execute(
                "select id, updated_at_ms, updated_at, archived from threads"
            ):
                fingerprints[thread_id] = f"{updated_ms or updated}|{int(bool(archived))}"
        finally:
            connection.close()
    units: dict[str, Unit] = {}
    for thread_id in sorted(set(rollouts) | set(fingerprints)):
        path = rollouts.get(thread_id)
        fingerprint = fingerprints.get(thread_id)
        if path is not None:
            units[thread_id] = Unit.from_path(thread_id, path, fingerprint)
        else:
            units[thread_id] = Unit(thread_id, None, fingerprint=fingerprint)
    return units

session_meta(thread_id, payload, path, fmt)

Merge a rollout's first line into the thread record; return the window anchor.

Source code in packages/agent-history/src/pdum/agent_history/codex.py

def session_meta(self, thread_id: str, payload: dict[str, Any], path: Path, fmt: str) -> datetime | None:
    """Merge a rollout's first line into the thread record; return the window anchor."""
    git = payload.get("git") or {}
    thread = self.threads.get(thread_id)
    if thread is None:
        thread = self.threads[thread_id] = {
            "thread_id": thread_id,
            "origin": "rollout-legacy" if fmt == "legacy" else "rollout",
            "archived": "archived_sessions" in path.parts,
        }
    thread.update(
        session_id=payload.get("session_id"),
        forked_from_id=payload.get("forked_from_id"),
        parent_thread_id=payload.get("parent_thread_id"),
        rollout_bytes=path.stat().st_size,
        rollout_format=fmt,
    )
    meta_time = schema.from_iso(payload.get("timestamp"))
    thread.setdefault("created_at", meta_time)
    thread.setdefault("cwd", payload.get("cwd"))
    thread.setdefault("originator", payload.get("originator"))
    thread.setdefault("cli_version", payload.get("cli_version"))
    thread.setdefault("model_provider", payload.get("model_provider"))
    thread.setdefault("history_mode", payload.get("history_mode"))
    for key, value in (
        ("git_branch", git.get("branch")),
        ("git_origin", git.get("repository_url")),
        ("git_sha", git.get("commit_hash")),
    ):
        if not thread.get(key) and value:
            thread[key] = value
    if not thread.get("repo"):
        thread["repo"] = repo_from_origin(thread.get("git_origin"))
    if payload.get("cli_version"):
        self.versions.add(str(payload["cli_version"]))
    thread.setdefault("rollout_path", str(path))
    if thread.get("source_kind") is None:
        source = payload.get("source")
        thread["source_kind"] = next(iter(source)) if isinstance(source, dict) and source else source
    thread.setdefault("thread_source", payload.get("thread_source"))
    thread.setdefault("agent_nickname", payload.get("agent_nickname"))
    thread.setdefault("agent_role", payload.get("agent_role"))
    anchor = thread.get("created_at") or meta_time
    if anchor is None:
        anchor = datetime.fromtimestamp(path.stat().st_mtime, tz=UTC).replace(tzinfo=None)
    return anchor

thread_in_window(thread_id, fallback_ms)

Decide membership for a projection row: only this run's units, by their window, else by timestamp.

Source code in packages/agent-history/src/pdum/agent_history/codex.py

def thread_in_window(self, thread_id: str, fallback_ms: Any) -> bool:
    """Decide membership for a projection row: only this run's units, by their window, else by timestamp."""
    if thread_id not in self.units:
        return False
    cover = self.coverage.get(thread_id)
    if cover is not None:
        return bool(cover["in_window"])
    if thread_id in self.window_by_thread:
        return self.window_by_thread[thread_id]
    return self.in_window(schema.from_ms(fallback_ms))

IngestSummary dataclass

What one ingest run read and wrote.

Parameters:

Name	Type	Description	Default
run_id	str	Identifier shared by the run's ingest_runs rows.	required
source	str	The source value written to every row.	required
source_id	str	Provenance key written to every row; stable for one agent, machine and root.	required
root	str	Directory that was read.	required
target	str	Where rows were published: a file path or md:<database>.	required
mode	str	rebuild for a fresh local file, sync for an incremental run, full for a sync that re-read every unit.	required
window_cutoff	str or None	ISO timestamp before which threads are only counted, or None when unlimited.	required

Source code in packages/agent-history/src/pdum/agent_history/base.py

@dataclass
class IngestSummary:
    """What one ingest run read and wrote.

    Parameters
    ----------
    run_id : str
        Identifier shared by the run's ``ingest_runs`` rows.
    source : str
        The ``source`` value written to every row.
    source_id : str
        Provenance key written to every row; stable for one agent, machine and root.
    root : str
        Directory that was read.
    target : str
        Where rows were published: a file path or ``md:<database>``.
    mode : str
        ``rebuild`` for a fresh local file, ``sync`` for an incremental run, ``full`` for a sync
        that re-read every unit.
    window_cutoff : str or None
        ISO timestamp before which threads are only counted, or None when unlimited.
    """

    run_id: str
    source: str
    source_id: str
    root: str
    target: str
    mode: str
    window_cutoff: str | None
    label: str = ""
    host: dict[str, str] = field(default_factory=dict)
    units: dict[str, int] = field(default_factory=dict)
    phases: dict[str, dict[str, Any]] = field(default_factory=dict)
    tables: dict[str, int] = field(default_factory=dict)
    derived: dict[str, Any] = field(default_factory=dict)
    unknown_line_types: dict[str, int] = field(default_factory=dict)
    ignored_line_types: dict[str, int] = field(default_factory=dict)
    notes: list[str] = field(default_factory=list)
    db_bytes: int = 0
    total_seconds: float = 0.0

    @property
    def db_path(self) -> str:
        """The target spec, kept under its old name for callers that printed it."""
        return self.target

    def to_dict(self) -> dict[str, Any]:
        """Return a JSON-compatible dictionary."""
        return asdict(self)

db_path property

The target spec, kept under its old name for callers that printed it.

to_dict()

Return a JSON-compatible dictionary.

Source code in packages/agent-history/src/pdum/agent_history/base.py

def to_dict(self) -> dict[str, Any]:
    """Return a JSON-compatible dictionary."""
    return asdict(self)

compact(db_path)

Rewrite a freshly written DuckDB file without the free blocks bulk inserts leave behind.

Parameters:

Name	Type	Description	Default
db_path	Path	Database to compact in place; a sibling temporary file is written and renamed over it.	required

Source code in packages/agent-history/src/pdum/agent_history/base.py

def compact(db_path: Path) -> None:
    """Rewrite a freshly written DuckDB file without the free blocks bulk inserts leave behind.

    Parameters
    ----------
    db_path : Path
        Database to compact in place; a sibling temporary file is written and renamed over it.
    """
    temporary = db_path.with_name(f".{db_path.name}.compact")
    temporary.unlink(missing_ok=True)
    connection = duckdb.connect(str(temporary))
    try:
        target = connection.execute("select current_database()").fetchone()[0]
        connection.execute(f"ATTACH '{db_path}' AS source (READ_ONLY)")
        connection.execute(f'COPY FROM DATABASE source TO "{target}"')
        connection.execute("CHECKPOINT")
    finally:
        connection.close()
    os.replace(temporary, db_path)

default_codex_home()

Return $CODEX_HOME or ~/.codex.

Source code in packages/agent-history/src/pdum/agent_history/codex.py

def default_codex_home() -> Path:
    """Return ``$CODEX_HOME`` or ``~/.codex``."""
    return Path(os.environ.get("CODEX_HOME") or Path.home() / ".codex").expanduser()

ingest_codex_home(home=None, db_path=None, *, window_days=DEFAULT_WINDOW_DAYS, blobs=True, now=None, mode='rebuild')

Ingest a Codex home into a DuckDB file or MotherDuck database.

Parameters:

Name	Type	Description	Default
home	Path or None	Codex home; defaults to $CODEX_HOME or ~/.codex.	None
db_path	Target or Path or str or None	Target DuckDB file or md:<database>; defaults to codex-history.duckdb in the current directory.	None
window_days	int or None	Support window in days; None ingests everything.	DEFAULT_WINDOW_DAYS
blobs	bool	Store full item JSON in item_blobs.	True
now	datetime or None	Reference time for the window, for tests.	None
mode	str	rebuild (fresh local file), sync (incremental) or full (sync re-reading everything).	'rebuild'

Returns:

Type	Description
IngestSummary	Phase timings, row counts and notes.

Source code in packages/agent-history/src/pdum/agent_history/codex.py

def ingest_codex_home(
    home: Path | None = None,
    db_path: Target | Path | str | None = None,
    *,
    window_days: int | None = DEFAULT_WINDOW_DAYS,
    blobs: bool = True,
    now: datetime | None = None,
    mode: str = "rebuild",
) -> IngestSummary:
    """Ingest a Codex home into a DuckDB file or MotherDuck database.

    Parameters
    ----------
    home : Path or None
        Codex home; defaults to ``$CODEX_HOME`` or ``~/.codex``.
    db_path : Target or Path or str or None
        Target DuckDB file or ``md:<database>``; defaults to ``codex-history.duckdb`` in the
        current directory.
    window_days : int or None
        Support window in days; None ingests everything.
    blobs : bool
        Store full item JSON in ``item_blobs``.
    now : datetime or None
        Reference time for the window, for tests.
    mode : str
        ``rebuild`` (fresh local file), ``sync`` (incremental) or ``full`` (sync re-reading everything).

    Returns
    -------
    IngestSummary
        Phase timings, row counts and notes.
    """
    ingest = CodexIngest(
        home or default_codex_home(),
        db_path or Path("codex-history.duckdb"),
        window_days=window_days,
        blobs=blobs,
        now=now,
        mode=mode,
    )
    return ingest.run()

Read a Claude Code configuration directory into the shared DuckDB schema.

Claude Code keeps one JSONL file per session under projects/<cwd-slug>/<session>.jsonl and one file per sub-agent under projects/<cwd-slug>/<session>/subagents/agent-<id>.jsonl. Lines form a tree through uuid and parentUuid; assistant lines carry the model, effort and a usage block per API response; tool calls and results are content blocks paired by id. There is no turn record, so turns are derived from user prompts. Sub-agents become their own threads with the session as parent. Older storage formats are counted, not modelled.

A continued or forked session copies its inherited transcript into the new file, ids and timestamps included. Since Claude Code 2.1.199 every copied message keeps the id of the session that produced it in session_id while sessionId names the file, so the marker decides what a file owns and names the direct predecessor (the last foreign producer before the file's own messages). Unmarked messages before a file's first own message are copies too; unmarked files are taken at face value. Sub-agents since 2.1.220 reference the parent's context (fork-context-ref) instead of copying it. Each file is one unit of incremental work: parsing it needs no other file, and what it says about other threads goes to thread_links for :mod:pdum.agent_history.derive to turn into fork edges and lineage after publishing.

METADATA_TYPES = {'agent-setting', 'mode', 'permission-mode', 'atis-latch', 'last-prompt', 'agent-name', 'relocated', 'worktree-state', 'file-history-snapshot', 'file-history-delta', 'api-request-blob', 'api-request-shape', 'artifact-autoreact-ledger', 'artifact-comment-monitor', 'bridge-session', 'frame-link', 'queue-operation'} module-attribute

Line types that carry no session facts we model; they are counted as ignored.

TOTAL_INDEX = [name for name, _ in schema.COLUMNS['responses']].index('total_tokens') module-attribute

Origin of lines a file copies from another session or from a parent before its own first prompt.

WORKFLOW_TYPES = {'launched', 'started', 'result'} module-attribute

Workflow-tool orchestration records inside subagents/workflows/ files.

ClaudeIngest

Bases: BaseIngest

One ingest of a Claude Code configuration directory into a fresh DuckDB file.

Parameters:

Name	Type	Description	Default
claude_dir	Path	Claude configuration directory holding projects/.	required
target	Target or Path or str	Local DuckDB file or md:<database> to publish into.	required
window_days	int or None	Sessions first seen before now - window_days are counted, not modelled. None disables.	DEFAULT_WINDOW_DAYS
blobs	bool	Store tool calls with their results and message blocks in item_blobs.	True
now	datetime or None	Reference time for the window; defaults to the current UTC time.	None
mode	str	rebuild, sync or full; see :class:~pdum.agent_history.base.BaseIngest.	'rebuild'

Source code in packages/agent-history/src/pdum/agent_history/claude.py

class ClaudeIngest(BaseIngest):
    """One ingest of a Claude Code configuration directory into a fresh DuckDB file.

    Parameters
    ----------
    claude_dir : Path
        Claude configuration directory holding ``projects/``.
    target : Target or Path or str
        Local DuckDB file or ``md:<database>`` to publish into.
    window_days : int or None
        Sessions first seen before ``now - window_days`` are counted, not modelled. None disables.
    blobs : bool
        Store tool calls with their results and message blocks in ``item_blobs``.
    now : datetime or None
        Reference time for the window; defaults to the current UTC time.
    mode : str
        ``rebuild``, ``sync`` or ``full``; see :class:`~pdum.agent_history.base.BaseIngest`.
    """

    def __init__(
        self,
        claude_dir: Path,
        target: Target | Path | str,
        *,
        window_days: int | None = DEFAULT_WINDOW_DAYS,
        blobs: bool = True,
        now: datetime | None = None,
        mode: str = "rebuild",
        host: dict[str, str] | None = None,
    ) -> None:
        super().__init__(
            SOURCE, "claude", claude_dir, target, window_days=window_days, blobs=blobs, now=now, host=host, mode=mode
        )
        self.threads: dict[str, _Thread] = {}

    # ------------------------------------------------------------------- scan
    def scan(self) -> dict[str, Unit]:
        """List every session and sub-agent file under ``projects/`` as a unit, from metadata alone."""
        projects = self.root / "projects"
        if not projects.is_dir():
            return {}
        units: dict[str, Unit] = {}
        for path in sorted(projects.glob("*/*.jsonl")):
            units[path.stem] = Unit.from_path(path.stem, path)
        for path in sorted(projects.glob("*/*/subagents/**/*.jsonl")):
            parts = path.relative_to(projects).parts
            session_id = parts[1]
            agent_id = path.stem.removeprefix("agent-")
            units[f"{session_id}:{agent_id}"] = Unit.from_path(f"{session_id}:{agent_id}", path, parent=session_id)
        return units

    # ----------------------------------------------------------------- ingest
    def ingest(self, units: dict[str, Unit]) -> None:
        """Read the given session and sub-agent files."""
        started = time.perf_counter()
        lines = 0
        skipped = 0
        mains = subs = 0
        for unit in sorted(units.values(), key=lambda u: str(u.path)):
            if unit.path is None:
                continue
            parent = unit.extra.get("parent")
            mains += parent is None
            subs += parent is not None
            before = set(self.threads)
            count = self.session(unit.path, parent)
            self.unit_threads[unit.unit_id] = sorted({unit.unit_id, *(set(self.threads) - before)})
            if count is None:
                skipped += 1
            else:
                lines += count
        self.phase(
            "sessions",
            started,
            lines,
            f"session files {mains}; sub-agent files {subs}; outside window {skipped}",
        )
        self.finish()

    # ---------------------------------------------------------------- session
    def session(self, path: Path, parent: str | None) -> int | None:
        """Read one session or sub-agent file; return its line count, or None when outside the window."""
        session_id = parent or path.stem
        agent_id = path.stem.removeprefix("agent-") if parent else None
        thread_id = f"{session_id}:{agent_id}" if parent else session_id
        anchor = self.anchor(path, session_id)
        cover = self.cover(thread_id)
        cover["rollout_format"] = FORMAT
        cover["in_window"] = self.in_window(anchor)
        thread = self.threads.setdefault(thread_id, _Thread(thread_id, parent=parent, path=path))
        thread.record["created_at"] = anchor
        thread.record["in_window"] = cover["in_window"]
        if parent:
            session_dir = next(p for p in path.parents if p.name == session_id)
            thread.record["agent_path"] = str(path.relative_to(session_dir).with_suffix(""))
            thread.record["agent_nickname"] = agent_id
        if not cover["in_window"]:
            cover["rollout_lines"] = None
            return None
        lines = 0
        with path.open("rb") as stream:
            for number, raw in enumerate(stream):
                lines += 1
                if raw.startswith(b'{"type":"api-request-blob"'):
                    self.ignored("api-request-blob")
                    continue
                try:
                    record = json.loads(raw)
                except ValueError:
                    cover["unknown_line_types"] += 1
                    self.unknown("unparseable")
                    continue
                self.line(thread, cover, record, number)
        cover["rollout_lines"] = lines
        self.close_turn(thread)
        return lines

    def anchor(self, path: Path, session_id: str) -> datetime:
        """Return when the session itself started, for the window decision.

        That is the timestamp of the first message the file did not inherit (no marker, or a
        marker naming this session), else the first timestamp in the file, else the modification
        time. A continued session carries its predecessor's old messages first, so the first
        timestamp alone would date it to its ancestor.
        """
        first: datetime | None = None
        with path.open("rb") as stream:
            for raw in stream:
                if b'"timestamp"' not in raw:
                    continue
                try:
                    record = json.loads(raw)
                except ValueError:
                    continue
                moment = schema.from_iso(record.get("timestamp"))
                if moment is None:
                    continue
                first = first or moment
                if record.get("type") in {"user", "assistant"}:
                    producer = record.get("session_id")
                    if producer is None or producer == session_id:
                        return moment
        if first is not None:
            return first
        return datetime.fromtimestamp(path.stat().st_mtime, tz=UTC).replace(tzinfo=None)

    # ------------------------------------------------------------------- line
    def line(self, thread: _Thread, cover: dict[str, Any], record: dict[str, Any], number: int) -> None:
        kind = record.get("type")
        ts = schema.from_iso(record.get("timestamp"))
        if record.get("isSidechain") and record.get("agentId") and thread.parent is None:
            # Older versions kept sub-agent lines inside the session file.
            child_id = f"{thread.thread_id}:{record['agentId']}"
            child = self.threads.get(child_id)
            if child is None:
                child = self.threads[child_id] = _Thread(child_id, parent=thread.thread_id, path=thread.path)
                child.record.update(created_at=ts, in_window=True, agent_path=f"agent-{record['agentId']}")
                child_cover = self.cover(child_id)
                child_cover["rollout_format"] = FORMAT
            thread, cover = child, self.cover(child_id)
        if kind != "continued-in":
            # A continued-in record is stamped when the successor starts, often days after the
            # last message here, so it neither ends the thread nor extends its last turn.
            thread.see(ts, activity=kind in {"user", "assistant", "attachment", "system"})
        self.describe_thread(thread, record)
        if kind in {"user", "assistant"}:
            thread.mark(record, ts)
            if thread.marker is not None:
                cover["marker_lines"] += 1
        if kind == "user":
            self.user(thread, cover, record, ts, number)
        elif kind == "assistant":
            self.assistant(thread, cover, record, ts, number)
        elif kind == "attachment":
            cover["response_items"] += 1
            rendered = record.get("rendered")
            text_len = len(rendered) if isinstance(rendered, str) else len(json.dumps(record.get("attachment") or {}))
            thread.response_items.append(
                schema.row(
                    "response_items",
                    thread_id=thread.thread_id,
                    turn_id=(thread.turn or {}).get("turn_id"),
                    ts=ts,
                    rollout_ordinal=number,
                    item_type="attachment",
                    role=record.get("renderedRole"),
                    text_len=text_len,
                    source=SOURCE,
                    origin="session-jsonl",
                )
            )
        elif kind == "system":
            self.system(thread, record, ts, number)
        elif kind == "cost-state":
            detail = {
                key: record.get(key)
                for key in (
                    "startTime",
                    "totalCostUSD",
                    "totalAPIDuration",
                    "totalAPIDurationWithoutRetries",
                    "totalToolDuration",
                    "totalLinesAdded",
                    "totalLinesRemoved",
                    "totalDuration",
                    "hasUnknownModelCost",
                    "modelUsage",
                )
            }
            self.event(thread, "cost-state", ts, number, detail)
        elif kind == "api-request":
            params = record.get("params") or {}
            detail = {
                "model": params.get("model"),
                "effort": (params.get("output_config") or {}).get("effort"),
                "max_tokens": params.get("max_tokens"),
                "cache_control_entries": len(record.get("cacheControl") or []),
                "query_source": record.get("querySource"),
            }
            self.event(thread, "api-request", ts, number, detail)
        elif kind == "fork-context-ref":
            parent_session = record.get("parentSessionId")
            if parent_session and parent_session != thread.thread_id:
                thread.record["forked_from_id"] = parent_session
            thread.fork_ref = record
            self.event(thread, "fork-context-ref", ts, number, {"context_length": record.get("contextLength")})
        elif kind == "continued-in":
            successor = record.get("continuedInSessionId")
            if isinstance(successor, str):
                thread.continued_in.append((successor, ts))
            self.event(thread, "continued-in", ts, number, {"session": successor})
        elif kind in {"ai-title", "custom-title"}:
            value = record.get("aiTitle") if kind == "ai-title" else record.get("customTitle")
            if isinstance(value, str):
                thread.titles[kind] = value
        elif kind in WORKFLOW_TYPES:
            detail = {key: record.get(key) for key in ("key", "label", "phase", "agentId") if key in record}
            self.event(thread, f"workflow:{kind}", ts, number, detail or None)
        elif kind in METADATA_TYPES:
            self.ignored(kind)
        else:
            cover["unknown_line_types"] += 1
            self.unknown(str(kind))

    def describe_thread(self, thread: _Thread, record: dict[str, Any]) -> None:
        record_fields = thread.record
        if record.get("cwd") and not record_fields.get("cwd"):
            record_fields["cwd"] = record["cwd"]
        if record.get("gitBranch") and not record_fields.get("git_branch"):
            record_fields["git_branch"] = record["gitBranch"]
        if record.get("version"):
            self.versions.add(str(record["version"]))
            if not record_fields.get("cli_version"):
                record_fields["cli_version"] = record["version"]
        if record.get("entrypoint") and not record_fields.get("originator"):
            record_fields["originator"] = record["entrypoint"]
        if record.get("sessionKind") and not record_fields.get("thread_source"):
            record_fields["thread_source"] = record["sessionKind"]

    # ------------------------------------------------------------------- user
    def user(self, thread: _Thread, cover: dict[str, Any], record: dict[str, Any], ts: Any, number: int) -> None:
        message = record.get("message") or {}
        content = message.get("content")
        blocks = content if isinstance(content, list) else [{"type": "text", "text": content}]
        blocks = [block for block in blocks if isinstance(block, dict)]
        if record.get("isCompactSummary"):
            cover["compactions"] += 1
            text, _ = _block_text(content)
            self.event(thread, "compacted", ts, number, {"summary_len": len(text)})
            item_id = record.get("uuid") or f"compact-{number}"
            thread.claim(item_id)
            thread.items[item_id] = {
                "thread_id": thread.thread_id,
                "turn_id": (thread.turn or {}).get("turn_id"),
                "item_id": item_id,
                "uuid": record.get("uuid"),
                "rollout_ordinal": number,
                "item_type": "contextCompaction",
                "created_at": ts,
                "completed_at": ts,
                "status": "completed",
                "text_len": len(text),
                "item_json_bytes": len(text),
            }
            return
        results = [block for block in blocks if block.get("type") == "tool_result"]
        if results:
            for block in results:
                self.tool_result(thread, block, record, ts)
            return
        text, images = _block_text(content)
        if record.get("isMeta") or not (text or images):
            cover["response_items"] += 1
            thread.response_items.append(
                schema.row(
                    "response_items",
                    thread_id=thread.thread_id,
                    turn_id=(thread.turn or {}).get("turn_id"),
                    ts=ts,
                    rollout_ordinal=number,
                    item_type="user-meta",
                    role="user",
                    text_len=len(text),
                    source=SOURCE,
                    origin="session-jsonl",
                )
            )
            return
        if text.lstrip().startswith(("<command-name>", "<local-command-", "<command-message>")):
            cover["response_items"] += 1
            thread.response_items.append(
                schema.row(
                    "response_items",
                    thread_id=thread.thread_id,
                    turn_id=(thread.turn or {}).get("turn_id"),
                    ts=ts,
                    rollout_ordinal=number,
                    item_type="local-command",
                    role="user",
                    text_len=len(text),
                    source=SOURCE,
                    origin="session-jsonl",
                )
            )
            return
        self.close_turn(thread)
        turn_id = record.get("promptId") or record.get("uuid") or f"turn-{number}"
        if turn_id in thread.turn_ids:
            turn_id = f"{turn_id}:{number}"
        thread.turn_ids.add(turn_id)
        copied = thread.marker == "foreign"
        if copied:
            thread.copied_ids.add(turn_id)
        elif thread.first_prompt_ts is None:
            thread.first_prompt_ts = ts
        thread.turns.append(
            {
                "thread_id": thread.thread_id,
                "turn_id": turn_id,
                "root_turn_id": turn_id,
                "rollout_ordinal": number,
                "started_at": ts,
                "last_ts": ts,
                "first_assistant_ts": None,
                "cwd": record.get("cwd"),
                "stop_reason": None,
                "api_error": False,
                "duration_from_system": None,
                "responses": set(),
                "origin": COPIED if copied else "session-jsonl",
            }
        )
        cover["turn_contexts"] += 1
        item_id = record.get("uuid") or f"user-{number}"
        thread.claim(item_id)
        thread.items[item_id] = {
            "thread_id": thread.thread_id,
            "turn_id": turn_id,
            "item_id": item_id,
            "uuid": record.get("uuid"),
            "rollout_ordinal": number,
            "item_type": "userMessage",
            "created_at": ts,
            "completed_at": ts,
            "status": "completed",
            "text": text[:20000],
            "text_len": len(text),
            "image_count": images,
            "n_parts": len(blocks),
            "item_json_bytes": len(json.dumps(content)) if content is not None else 0,
        }
        if self.blobs:
            thread.blobs[item_id] = {"type": "userMessage", "content": content}

    # -------------------------------------------------------------- assistant
    def assistant(self, thread: _Thread, cover: dict[str, Any], record: dict[str, Any], ts: Any, number: int) -> None:
        message = record.get("message") or {}
        if thread.turn is None:
            thread.turns.append(
                {
                    "thread_id": thread.thread_id,
                    "turn_id": f"implicit-{record.get('uuid') or number}",
                    "root_turn_id": None,
                    "rollout_ordinal": number,
                    "started_at": ts,
                    "last_ts": ts,
                    "first_assistant_ts": None,
                    "cwd": record.get("cwd"),
                    "stop_reason": None,
                    "api_error": False,
                    "duration_from_system": None,
                    "responses": set(),
                    "origin": COPIED,
                }
            )
        turn = thread.turn
        assert turn is not None
        if turn["first_assistant_ts"] is None:
            turn["first_assistant_ts"] = ts
        model = message.get("model")
        effort = record.get("effort") or record.get("perTurnEffort")
        if isinstance(model, str):
            thread.models[model] += 1
            turn["model"] = model
        if isinstance(effort, str):
            thread.efforts[effort] += 1
            turn["reasoning_effort"] = effort
        if record.get("isApiErrorMessage") or record.get("apiErrorStatus"):
            turn["api_error"] = True
            self.event(
                thread,
                "api-error",
                ts,
                number,
                {"status": record.get("apiErrorStatus"), "error": str(record.get("error"))[:200]},
            )
        if message.get("stop_reason"):
            turn["stop_reason"] = message["stop_reason"]
        if record.get("isAbortedMidStream"):
            self.event(thread, "aborted-mid-stream", ts, number, {"model": model})
        response_id = message.get("id") or record.get("requestId") or record.get("uuid") or f"resp-{number}"
        thread.claim(response_id)
        response = thread.responses.get(response_id)
        if response is None:
            response = thread.responses[response_id] = {
                "response_id": response_id,
                "turn_id": turn["turn_id"],
                "ts": ts,
                "rollout_ordinal": number,
                "model": model,
                "usage": None,
                "producer": thread.producer,
            }
            cover["usage_records"] += 1
        usage = message.get("usage")
        if isinstance(usage, dict):
            # One API response is written as one record per content block. Sub-agent files put a
            # placeholder output count on earlier blocks and the real one on the last, and a fork copy
            # can carry a zero with the same timestamp, so keep the member with the most output.
            current = response.get("usage")
            if current is None or int(usage.get("output_tokens") or 0) >= int(current.get("output_tokens") or 0):
                response["usage"] = usage
        turn["responses"].add(response_id)
        content = message.get("content")
        blocks = content if isinstance(content, list) else [{"type": "text", "text": content}]
        for index, block in enumerate(blocks):
            if not isinstance(block, dict):
                continue
            kind = block.get("type")
            item_id = (
                f"{record.get('uuid') or number}:{index}" if kind != "tool_use" else block.get("id") or f"tool-{number}"
            )
            thread.claim(item_id)
            base = {
                "thread_id": thread.thread_id,
                "turn_id": turn["turn_id"],
                "item_id": item_id,
                "uuid": record.get("uuid"),
                "rollout_ordinal": number,
                "created_at": ts,
                "status": "completed",
                "cwd": record.get("cwd"),
                "item_json_bytes": len(json.dumps(block)),
            }
            if kind == "text":
                text = block.get("text") or ""
                base.update(
                    item_type="agentMessage",
                    completed_at=ts,
                    text=text[:20000],
                    text_len=len(text),
                    has_mermaid="```mermaid" in text,
                    image_count=text.count("!["),
                )
                thread.items[item_id] = base
                if self.blobs:
                    thread.blobs[item_id] = {"type": "agentMessage", "block": block}
            elif kind == "thinking":
                thinking = block.get("thinking") or ""
                base.update(
                    item_type="reasoning",
                    completed_at=ts,
                    text_len=len(thinking),
                    exec_duration_ms=record.get("thinkingDurationMs"),
                )
                thread.items[item_id] = base
            elif kind == "tool_use":
                base.update(self.tool_use(block, record))
                base.update(item_type=base.get("item_type"), started_at=ts, status="inProgress")
                thread.items[item_id] = base
                if self.blobs:
                    thread.blobs[item_id] = {"type": "toolCall", "call": block, "result": None, "toolUseResult": None}
            else:
                self.ignored(f"assistant:{kind}")

    def tool_use(self, block: dict[str, Any], record: dict[str, Any]) -> dict[str, Any]:
        """Map one Claude ``tool_use`` block onto item columns."""
        name = block.get("name") or ""
        inputs = block.get("input") if isinstance(block.get("input"), dict) else {}
        out: dict[str, Any] = {"tool": name, "exec_source": "claude"}
        if name == "Bash":
            command = inputs.get("command") or ""
            out.update(
                item_type="commandExecution",
                command=command[:4000],
                command_program=item_columns.program(command),
                text=command[:20000],
                text_len=len(command),
                agent_kind="background" if inputs.get("run_in_background") else None,
            )
        elif name in EDIT_TOOLS:
            path = inputs.get("file_path") or inputs.get("notebook_path")
            added = deleted = 0
            if name == "Write":
                added = _lines(inputs.get("content"))
            elif name == "MultiEdit":
                for edit in inputs.get("edits") or []:
                    if isinstance(edit, dict):
                        added += _lines(edit.get("new_string"))
                        deleted += _lines(edit.get("old_string"))
            else:
                added = _lines(inputs.get("new_string") or inputs.get("new_source"))
                deleted = _lines(inputs.get("old_string"))
            out.update(
                item_type="fileChange",
                paths=json.dumps([path]),
                n_parts=1,
                added_lines=added,
                deleted_lines=deleted,
                text=path,
                text_len=len(path) if isinstance(path, str) else None,
            )
        elif name.startswith("mcp__"):
            parts = name.split("__", 2)
            server = parts[1] if len(parts) > 1 else None
            tool = parts[2] if len(parts) > 2 else name
            out.update(item_type="mcpToolCall", server=server, tool=tool, text=f"{server}:{tool}", text_len=len(name))
        elif name == "Agent":
            description = inputs.get("description") or inputs.get("prompt") or ""
            out.update(
                item_type="collabAgentToolCall",
                agent_kind=inputs.get("subagent_type") or "agent",
                agent_path=inputs.get("name"),
                text=description[:20000],
                text_len=len(description),
            )
        elif name in SEARCH_TOOLS:
            query = inputs.get("query") or inputs.get("url") or ""
            out.update(item_type="webSearch", text=query[:20000], text_len=len(query))
        else:
            summary = next((inputs.get(key) for key in SUMMARY_KEYS if isinstance(inputs.get(key), str)), None)
            out.update(item_type="toolCall", text=summary, text_len=len(summary) if summary else None)
        return out

    def tool_result(self, thread: _Thread, block: dict[str, Any], record: dict[str, Any], ts: Any) -> None:
        """Complete the item a ``tool_result`` block answers."""
        item = thread.items.get(block.get("tool_use_id") or "")
        structured = record.get("toolUseResult")
        if item is None:
            self.ignored("orphan-tool-result")
            return
        text, images = _block_text(block.get("content"))
        started = item.get("started_at")
        item["completed_at"] = ts
        if started is not None and ts is not None:
            item["duration_ms"] = int((ts - started).total_seconds() * 1000)
        item["status"] = "failed" if block.get("is_error") else "completed"
        item["output_bytes"] = len(text.encode())
        item["image_count"] = (item.get("image_count") or 0) + images
        if isinstance(structured, dict):
            if "stdout" in structured or "stderr" in structured:
                output = (structured.get("stdout") or "") + (structured.get("stderr") or "")
                item["output_bytes"] = len(output.encode())
                item["output_truncated"] = bool(structured.get("persistedOutputPath"))
                if structured.get("interrupted"):
                    item["status"] = "interrupted"
            counts = patch_line_counts(structured.get("structuredPatch"))
            if counts is not None and item.get("item_type") == "fileChange":
                item["added_lines"], item["deleted_lines"] = counts
        if item.get("item_type") == "mcpToolCall":
            item["mcp_error"] = bool(block.get("is_error"))
        if self.blobs and item["item_id"] in thread.blobs:
            blob = thread.blobs[item["item_id"]]
            blob["result"] = block
            blob["toolUseResult"] = structured

    # ----------------------------------------------------------------- system
    def system(self, thread: _Thread, record: dict[str, Any], ts: Any, number: int) -> None:
        subtype = record.get("subtype") or "unknown"
        detail = {
            key: record.get(key) for key in ("level", "durationMs", "messageCount", "compactMetadata") if key in record
        }
        if subtype == "turn_duration" and thread.turn is not None and record.get("durationMs") is not None:
            thread.turn["duration_from_system"] = record["durationMs"]
        self.event(thread, f"system:{subtype}", ts, number, detail)

    def event(self, thread: _Thread, kind: str, ts: Any, number: int, detail: dict[str, Any] | None) -> None:
        thread.events.append(
            schema.row(
                "rollout_events",
                thread_id=thread.thread_id,
                turn_id=(thread.turn or {}).get("turn_id"),
                ts=ts,
                rollout_ordinal=number,
                event_type=kind,
                detail=json.dumps(detail, default=str) if detail else None,
                source=SOURCE,
                origin="session-jsonl",
            )
        )

    def close_turn(self, thread: _Thread) -> None:
        """Finalize the open turn's status and timing."""
        turn = thread.turn
        if turn is None or turn.get("closed"):
            return
        turn["closed"] = True
        if turn["api_error"]:
            turn["status"] = "failed"
        elif turn["stop_reason"] in {"end_turn", "stop_sequence", "max_tokens"}:
            turn["status"] = "completed"
        elif turn["responses"]:
            turn["status"] = "interrupted"
        else:
            turn["status"] = "noResponse"
        started = turn.get("started_at")
        last = turn.get("last_ts")
        turn["completed_at"] = last
        if turn["duration_from_system"] is not None:
            turn["duration_ms"] = turn["duration_from_system"]
        elif started is not None and last is not None:
            turn["duration_ms"] = int((last - started).total_seconds() * 1000)
        first = turn.get("first_assistant_ts")
        if started is not None and first is not None:
            turn["time_to_first_token_ms"] = int((first - started).total_seconds() * 1000)

    # ----------------------------------------------------------------- finish
    def finish(self) -> None:
        started = time.perf_counter()
        thread_columns = dict(schema.COLUMNS["threads"])
        turn_columns = dict(schema.COLUMNS["turns"])
        item_columns_set = dict(schema.COLUMNS["items"])
        thread_rows: list[tuple] = []
        turn_rows: list[tuple] = []
        for thread in self.threads.values():
            self.close_turn(thread)
            for turn in thread.turns:
                responses = turn["responses"]
                inherited_prompt = turn["turn_id"] in thread.copied_ids
                if responses:
                    copied_turn = all(
                        thread.is_copied(None, r.get("ts"), rid)
                        for rid, r in thread.responses.items()
                        if rid in responses
                    )
                else:
                    copied_turn = inherited_prompt
                turn["origin"] = COPIED if copied_turn else "session-jsonl"
            record = thread.record
            record.setdefault("in_window", True)
            own_start = self.own_start(thread)
            if own_start is not None:
                record["created_at"] = own_start
            record["updated_at"] = thread.last_ts
            record["title"] = thread.titles.get("custom-title") or thread.titles.get("ai-title")
            record["model"] = thread.models.most_common(1)[0][0] if thread.models else None
            record["reasoning_effort"] = thread.efforts.most_common(1)[0][0] if thread.efforts else None
            record["source"] = SOURCE
            if not record.get("in_window"):
                thread_rows.append(schema.row("threads", **{k: v for k, v in record.items() if k in thread_columns}))
                continue
            responses, counts = self.usage_rows(thread)
            record["tokens_used"] = sum(row[TOTAL_INDEX] for row in responses) if responses else 0
            thread_rows.append(schema.row("threads", **{k: v for k, v in record.items() if k in thread_columns}))
            for turn in thread.turns:
                turn_rows.append(
                    schema.row("turns", **{k: v for k, v in turn.items() if k in turn_columns}, source=SOURCE)
                )
            self.write("responses", responses)
            self.write("token_counts", counts)
            self.write(
                "items",
                [
                    schema.row(
                        "items",
                        **{k: v for k, v in item.items() if k in item_columns_set},
                        source=SOURCE,
                        origin=COPIED
                        if thread.is_copied(item.get("turn_id"), item.get("created_at"), item["item_id"])
                        else "session-jsonl",
                    )
                    for item in thread.items.values()
                ],
            )
            if self.blobs:
                self.write(
                    "item_blobs",
                    [
                        schema.row(
                            "item_blobs",
                            thread_id=thread.thread_id,
                            turn_id=thread.items[item_id]["turn_id"],
                            item_id=item_id,
                            item_json=json.dumps(blob, default=str),
                            source=SOURCE,
                            origin="session-jsonl",
                        )
                        for item_id, blob in thread.blobs.items()
                        if item_id in thread.items
                    ],
                )
            self.write("rollout_events", thread.events)
            self.write("response_items", thread.response_items)
            self.write("thread_links", self.links(thread))
            if thread.parent:
                self.write(
                    "spawn_edges",
                    [
                        schema.row(
                            "spawn_edges",
                            parent_thread_id=thread.parent,
                            child_thread_id=thread.thread_id,
                            status="recorded",
                            source=SOURCE,
                            origin="session-jsonl",
                        )
                    ],
                )
        self.write("threads", thread_rows)
        self.write("turns", turn_rows)
        cover_rows = self.write_coverage_and_meta()
        self.phase(
            "finish",
            started,
            len(thread_rows) + len(turn_rows) + cover_rows,
            f"threads {len(thread_rows)}; turns {len(turn_rows)}; coverage {cover_rows}",
        )

    def own_start(self, thread: _Thread) -> datetime | None:
        """Return when the thread produced its first own message, from markers or the boundary rule."""
        if thread.own_first_ts is not None:
            return thread.own_first_ts
        moments = [
            r["ts"] for key, r in thread.responses.items() if r.get("ts") and not thread.is_copied(None, r["ts"], key)
        ]
        moments += [
            item["created_at"]
            for key, item in thread.items.items()
            if item.get("created_at") and not thread.is_copied(None, item["created_at"], key)
        ]
        return min(moments) if moments else None

    def links(self, thread: _Thread) -> list[tuple]:
        """What this file says about other threads: producers it copied, its successor, its fork reference."""
        rows: list[tuple] = []
        for ordinal, (producer, count) in enumerate(thread.producers.items()):
            rows.append(
                schema.row(
                    "thread_links",
                    thread_id=thread.thread_id,
                    kind="producer",
                    target_thread_id=producer,
                    ordinal=ordinal,
                    records=count,
                    ts=thread.producer_last_ts.get(producer),
                    source=SOURCE,
                    origin="session-jsonl",
                )
            )
        for ordinal, (successor, ts) in enumerate(thread.continued_in):
            rows.append(
                schema.row(
                    "thread_links",
                    thread_id=thread.thread_id,
                    kind="continued-in",
                    target_thread_id=successor,
                    ordinal=ordinal,
                    ts=ts,
                    source=SOURCE,
                    origin="session-jsonl",
                )
            )
        if thread.fork_ref is not None:
            ref = thread.fork_ref
            rows.append(
                schema.row(
                    "thread_links",
                    thread_id=thread.thread_id,
                    kind="fork-context-ref",
                    target_thread_id=ref.get("parentSessionId") or thread.parent,
                    ordinal=0,
                    records=ref.get("contextLength"),
                    ts=thread.first_prompt_ts or thread.first_ts,
                    detail=json.dumps({"parentLastUuid": ref.get("parentLastUuid")}),
                    source=SOURCE,
                    origin="session-jsonl",
                )
            )
        return rows

    def usage_rows(self, thread: _Thread) -> tuple[list[tuple], list[tuple]]:
        """Build ``responses`` and synthesized ``token_counts`` rows with running totals."""
        responses: list[tuple] = []
        counts: list[tuple] = []
        thread_totals = {"input": 0, "cached": 0, "cache_write": 0, "output": 0, "total": 0}
        turn_totals: dict[str, dict[str, int]] = {}
        for response in thread.responses.values():
            copied = thread.is_copied(response["turn_id"], response.get("ts"), response["response_id"])
            usage = response.get("usage") or {}
            cached = int(usage.get("cache_read_input_tokens") or 0)
            cache_write = int(usage.get("cache_creation_input_tokens") or 0)
            tiers = usage.get("cache_creation") if isinstance(usage.get("cache_creation"), dict) else {}
            cache_write_1h = int(tiers.get("ephemeral_1h_input_tokens") or 0)
            uncached = int(usage.get("input_tokens") or 0)
            output = int(usage.get("output_tokens") or 0)
            iterations = usage.get("iterations") if isinstance(usage.get("iterations"), list) else []
            attempts = sum(int((it or {}).get("output_tokens") or 0) for it in iterations if isinstance(it, dict))
            wasted = max(attempts - output, 0) if len(iterations) > 1 else 0
            total_input = uncached + cached + cache_write
            total = total_input + output
            turn = turn_totals.setdefault(response["turn_id"], {"input": 0, "cached": 0, "output": 0, "total": 0})
            if not copied:
                for bucket, value in (("input", total_input), ("cached", cached), ("output", output), ("total", total)):
                    thread_totals[bucket] += value
                    turn[bucket] += value
                thread_totals["cache_write"] += cache_write
                self.cover(thread.thread_id)["token_counts"] += 1
            responses.append(
                schema.row(
                    "responses",
                    thread_id=thread.thread_id,
                    turn_id=response["turn_id"],
                    root_turn_id=response["turn_id"],
                    session_id=response.get("producer") or thread.parent or thread.thread_id,
                    response_id=response["response_id"],
                    ts=response["ts"],
                    rollout_ordinal=response["rollout_ordinal"],
                    model=response.get("model"),
                    input_tokens=total_input,
                    cached_input_tokens=cached,
                    cache_write_input_tokens=cache_write,
                    cache_write_1h_input_tokens=cache_write_1h,
                    output_tokens=output,
                    reasoning_output_tokens=None,
                    wasted_output_tokens=wasted,
                    total_tokens=total,
                    turn_input_tokens=turn["input"],
                    turn_cached_input_tokens=turn["cached"],
                    turn_output_tokens=turn["output"],
                    turn_total_tokens=turn["total"],
                    thread_input_tokens=thread_totals["input"],
                    thread_cached_input_tokens=thread_totals["cached"],
                    thread_output_tokens=thread_totals["output"],
                    thread_total_tokens=thread_totals["total"],
                    source=SOURCE,
                    origin=COPIED if copied else "assistant",
                )
            )
            if copied:
                continue
            counts.append(
                schema.row(
                    "token_counts",
                    thread_id=thread.thread_id,
                    turn_id=response["turn_id"],
                    ts=response["ts"],
                    rollout_ordinal=response["rollout_ordinal"],
                    total_input_tokens=thread_totals["input"],
                    total_cached_input_tokens=thread_totals["cached"],
                    total_cache_write_input_tokens=thread_totals["cache_write"],
                    total_output_tokens=thread_totals["output"],
                    total_tokens=thread_totals["total"],
                    last_input_tokens=total_input,
                    last_cached_input_tokens=cached,
                    last_cache_write_input_tokens=cache_write,
                    last_output_tokens=output,
                    last_total_tokens=total,
                    source=SOURCE,
                    origin="assistant",
                )
            )
        return responses, counts

anchor(path, session_id)

Return when the session itself started, for the window decision.

That is the timestamp of the first message the file did not inherit (no marker, or a marker naming this session), else the first timestamp in the file, else the modification time. A continued session carries its predecessor's old messages first, so the first timestamp alone would date it to its ancestor.

Source code in packages/agent-history/src/pdum/agent_history/claude.py

def anchor(self, path: Path, session_id: str) -> datetime:
    """Return when the session itself started, for the window decision.

    That is the timestamp of the first message the file did not inherit (no marker, or a
    marker naming this session), else the first timestamp in the file, else the modification
    time. A continued session carries its predecessor's old messages first, so the first
    timestamp alone would date it to its ancestor.
    """
    first: datetime | None = None
    with path.open("rb") as stream:
        for raw in stream:
            if b'"timestamp"' not in raw:
                continue
            try:
                record = json.loads(raw)
            except ValueError:
                continue
            moment = schema.from_iso(record.get("timestamp"))
            if moment is None:
                continue
            first = first or moment
            if record.get("type") in {"user", "assistant"}:
                producer = record.get("session_id")
                if producer is None or producer == session_id:
                    return moment
    if first is not None:
        return first
    return datetime.fromtimestamp(path.stat().st_mtime, tz=UTC).replace(tzinfo=None)

close_turn(thread)

Finalize the open turn's status and timing.

Source code in packages/agent-history/src/pdum/agent_history/claude.py

def close_turn(self, thread: _Thread) -> None:
    """Finalize the open turn's status and timing."""
    turn = thread.turn
    if turn is None or turn.get("closed"):
        return
    turn["closed"] = True
    if turn["api_error"]:
        turn["status"] = "failed"
    elif turn["stop_reason"] in {"end_turn", "stop_sequence", "max_tokens"}:
        turn["status"] = "completed"
    elif turn["responses"]:
        turn["status"] = "interrupted"
    else:
        turn["status"] = "noResponse"
    started = turn.get("started_at")
    last = turn.get("last_ts")
    turn["completed_at"] = last
    if turn["duration_from_system"] is not None:
        turn["duration_ms"] = turn["duration_from_system"]
    elif started is not None and last is not None:
        turn["duration_ms"] = int((last - started).total_seconds() * 1000)
    first = turn.get("first_assistant_ts")
    if started is not None and first is not None:
        turn["time_to_first_token_ms"] = int((first - started).total_seconds() * 1000)

ingest(units)

Read the given session and sub-agent files.

Source code in packages/agent-history/src/pdum/agent_history/claude.py

def ingest(self, units: dict[str, Unit]) -> None:
    """Read the given session and sub-agent files."""
    started = time.perf_counter()
    lines = 0
    skipped = 0
    mains = subs = 0
    for unit in sorted(units.values(), key=lambda u: str(u.path)):
        if unit.path is None:
            continue
        parent = unit.extra.get("parent")
        mains += parent is None
        subs += parent is not None
        before = set(self.threads)
        count = self.session(unit.path, parent)
        self.unit_threads[unit.unit_id] = sorted({unit.unit_id, *(set(self.threads) - before)})
        if count is None:
            skipped += 1
        else:
            lines += count
    self.phase(
        "sessions",
        started,
        lines,
        f"session files {mains}; sub-agent files {subs}; outside window {skipped}",
    )
    self.finish()

links(thread)

What this file says about other threads: producers it copied, its successor, its fork reference.

Source code in packages/agent-history/src/pdum/agent_history/claude.py

def links(self, thread: _Thread) -> list[tuple]:
    """What this file says about other threads: producers it copied, its successor, its fork reference."""
    rows: list[tuple] = []
    for ordinal, (producer, count) in enumerate(thread.producers.items()):
        rows.append(
            schema.row(
                "thread_links",
                thread_id=thread.thread_id,
                kind="producer",
                target_thread_id=producer,
                ordinal=ordinal,
                records=count,
                ts=thread.producer_last_ts.get(producer),
                source=SOURCE,
                origin="session-jsonl",
            )
        )
    for ordinal, (successor, ts) in enumerate(thread.continued_in):
        rows.append(
            schema.row(
                "thread_links",
                thread_id=thread.thread_id,
                kind="continued-in",
                target_thread_id=successor,
                ordinal=ordinal,
                ts=ts,
                source=SOURCE,
                origin="session-jsonl",
            )
        )
    if thread.fork_ref is not None:
        ref = thread.fork_ref
        rows.append(
            schema.row(
                "thread_links",
                thread_id=thread.thread_id,
                kind="fork-context-ref",
                target_thread_id=ref.get("parentSessionId") or thread.parent,
                ordinal=0,
                records=ref.get("contextLength"),
                ts=thread.first_prompt_ts or thread.first_ts,
                detail=json.dumps({"parentLastUuid": ref.get("parentLastUuid")}),
                source=SOURCE,
                origin="session-jsonl",
            )
        )
    return rows

own_start(thread)

Return when the thread produced its first own message, from markers or the boundary rule.

Source code in packages/agent-history/src/pdum/agent_history/claude.py

def own_start(self, thread: _Thread) -> datetime | None:
    """Return when the thread produced its first own message, from markers or the boundary rule."""
    if thread.own_first_ts is not None:
        return thread.own_first_ts
    moments = [
        r["ts"] for key, r in thread.responses.items() if r.get("ts") and not thread.is_copied(None, r["ts"], key)
    ]
    moments += [
        item["created_at"]
        for key, item in thread.items.items()
        if item.get("created_at") and not thread.is_copied(None, item["created_at"], key)
    ]
    return min(moments) if moments else None

scan()

List every session and sub-agent file under projects/ as a unit, from metadata alone.

Source code in packages/agent-history/src/pdum/agent_history/claude.py

def scan(self) -> dict[str, Unit]:
    """List every session and sub-agent file under ``projects/`` as a unit, from metadata alone."""
    projects = self.root / "projects"
    if not projects.is_dir():
        return {}
    units: dict[str, Unit] = {}
    for path in sorted(projects.glob("*/*.jsonl")):
        units[path.stem] = Unit.from_path(path.stem, path)
    for path in sorted(projects.glob("*/*/subagents/**/*.jsonl")):
        parts = path.relative_to(projects).parts
        session_id = parts[1]
        agent_id = path.stem.removeprefix("agent-")
        units[f"{session_id}:{agent_id}"] = Unit.from_path(f"{session_id}:{agent_id}", path, parent=session_id)
    return units

session(path, parent)

Read one session or sub-agent file; return its line count, or None when outside the window.

Source code in packages/agent-history/src/pdum/agent_history/claude.py

def session(self, path: Path, parent: str | None) -> int | None:
    """Read one session or sub-agent file; return its line count, or None when outside the window."""
    session_id = parent or path.stem
    agent_id = path.stem.removeprefix("agent-") if parent else None
    thread_id = f"{session_id}:{agent_id}" if parent else session_id
    anchor = self.anchor(path, session_id)
    cover = self.cover(thread_id)
    cover["rollout_format"] = FORMAT
    cover["in_window"] = self.in_window(anchor)
    thread = self.threads.setdefault(thread_id, _Thread(thread_id, parent=parent, path=path))
    thread.record["created_at"] = anchor
    thread.record["in_window"] = cover["in_window"]
    if parent:
        session_dir = next(p for p in path.parents if p.name == session_id)
        thread.record["agent_path"] = str(path.relative_to(session_dir).with_suffix(""))
        thread.record["agent_nickname"] = agent_id
    if not cover["in_window"]:
        cover["rollout_lines"] = None
        return None
    lines = 0
    with path.open("rb") as stream:
        for number, raw in enumerate(stream):
            lines += 1
            if raw.startswith(b'{"type":"api-request-blob"'):
                self.ignored("api-request-blob")
                continue
            try:
                record = json.loads(raw)
            except ValueError:
                cover["unknown_line_types"] += 1
                self.unknown("unparseable")
                continue
            self.line(thread, cover, record, number)
    cover["rollout_lines"] = lines
    self.close_turn(thread)
    return lines

tool_result(thread, block, record, ts)

Complete the item a tool_result block answers.

Source code in packages/agent-history/src/pdum/agent_history/claude.py

def tool_result(self, thread: _Thread, block: dict[str, Any], record: dict[str, Any], ts: Any) -> None:
    """Complete the item a ``tool_result`` block answers."""
    item = thread.items.get(block.get("tool_use_id") or "")
    structured = record.get("toolUseResult")
    if item is None:
        self.ignored("orphan-tool-result")
        return
    text, images = _block_text(block.get("content"))
    started = item.get("started_at")
    item["completed_at"] = ts
    if started is not None and ts is not None:
        item["duration_ms"] = int((ts - started).total_seconds() * 1000)
    item["status"] = "failed" if block.get("is_error") else "completed"
    item["output_bytes"] = len(text.encode())
    item["image_count"] = (item.get("image_count") or 0) + images
    if isinstance(structured, dict):
        if "stdout" in structured or "stderr" in structured:
            output = (structured.get("stdout") or "") + (structured.get("stderr") or "")
            item["output_bytes"] = len(output.encode())
            item["output_truncated"] = bool(structured.get("persistedOutputPath"))
            if structured.get("interrupted"):
                item["status"] = "interrupted"
        counts = patch_line_counts(structured.get("structuredPatch"))
        if counts is not None and item.get("item_type") == "fileChange":
            item["added_lines"], item["deleted_lines"] = counts
    if item.get("item_type") == "mcpToolCall":
        item["mcp_error"] = bool(block.get("is_error"))
    if self.blobs and item["item_id"] in thread.blobs:
        blob = thread.blobs[item["item_id"]]
        blob["result"] = block
        blob["toolUseResult"] = structured

tool_use(block, record)

Map one Claude tool_use block onto item columns.

Source code in packages/agent-history/src/pdum/agent_history/claude.py

def tool_use(self, block: dict[str, Any], record: dict[str, Any]) -> dict[str, Any]:
    """Map one Claude ``tool_use`` block onto item columns."""
    name = block.get("name") or ""
    inputs = block.get("input") if isinstance(block.get("input"), dict) else {}
    out: dict[str, Any] = {"tool": name, "exec_source": "claude"}
    if name == "Bash":
        command = inputs.get("command") or ""
        out.update(
            item_type="commandExecution",
            command=command[:4000],
            command_program=item_columns.program(command),
            text=command[:20000],
            text_len=len(command),
            agent_kind="background" if inputs.get("run_in_background") else None,
        )
    elif name in EDIT_TOOLS:
        path = inputs.get("file_path") or inputs.get("notebook_path")
        added = deleted = 0
        if name == "Write":
            added = _lines(inputs.get("content"))
        elif name == "MultiEdit":
            for edit in inputs.get("edits") or []:
                if isinstance(edit, dict):
                    added += _lines(edit.get("new_string"))
                    deleted += _lines(edit.get("old_string"))
        else:
            added = _lines(inputs.get("new_string") or inputs.get("new_source"))
            deleted = _lines(inputs.get("old_string"))
        out.update(
            item_type="fileChange",
            paths=json.dumps([path]),
            n_parts=1,
            added_lines=added,
            deleted_lines=deleted,
            text=path,
            text_len=len(path) if isinstance(path, str) else None,
        )
    elif name.startswith("mcp__"):
        parts = name.split("__", 2)
        server = parts[1] if len(parts) > 1 else None
        tool = parts[2] if len(parts) > 2 else name
        out.update(item_type="mcpToolCall", server=server, tool=tool, text=f"{server}:{tool}", text_len=len(name))
    elif name == "Agent":
        description = inputs.get("description") or inputs.get("prompt") or ""
        out.update(
            item_type="collabAgentToolCall",
            agent_kind=inputs.get("subagent_type") or "agent",
            agent_path=inputs.get("name"),
            text=description[:20000],
            text_len=len(description),
        )
    elif name in SEARCH_TOOLS:
        query = inputs.get("query") or inputs.get("url") or ""
        out.update(item_type="webSearch", text=query[:20000], text_len=len(query))
    else:
        summary = next((inputs.get(key) for key in SUMMARY_KEYS if isinstance(inputs.get(key), str)), None)
        out.update(item_type="toolCall", text=summary, text_len=len(summary) if summary else None)
    return out

usage_rows(thread)

Build responses and synthesized token_counts rows with running totals.

Source code in packages/agent-history/src/pdum/agent_history/claude.py

def usage_rows(self, thread: _Thread) -> tuple[list[tuple], list[tuple]]:
    """Build ``responses`` and synthesized ``token_counts`` rows with running totals."""
    responses: list[tuple] = []
    counts: list[tuple] = []
    thread_totals = {"input": 0, "cached": 0, "cache_write": 0, "output": 0, "total": 0}
    turn_totals: dict[str, dict[str, int]] = {}
    for response in thread.responses.values():
        copied = thread.is_copied(response["turn_id"], response.get("ts"), response["response_id"])
        usage = response.get("usage") or {}
        cached = int(usage.get("cache_read_input_tokens") or 0)
        cache_write = int(usage.get("cache_creation_input_tokens") or 0)
        tiers = usage.get("cache_creation") if isinstance(usage.get("cache_creation"), dict) else {}
        cache_write_1h = int(tiers.get("ephemeral_1h_input_tokens") or 0)
        uncached = int(usage.get("input_tokens") or 0)
        output = int(usage.get("output_tokens") or 0)
        iterations = usage.get("iterations") if isinstance(usage.get("iterations"), list) else []
        attempts = sum(int((it or {}).get("output_tokens") or 0) for it in iterations if isinstance(it, dict))
        wasted = max(attempts - output, 0) if len(iterations) > 1 else 0
        total_input = uncached + cached + cache_write
        total = total_input + output
        turn = turn_totals.setdefault(response["turn_id"], {"input": 0, "cached": 0, "output": 0, "total": 0})
        if not copied:
            for bucket, value in (("input", total_input), ("cached", cached), ("output", output), ("total", total)):
                thread_totals[bucket] += value
                turn[bucket] += value
            thread_totals["cache_write"] += cache_write
            self.cover(thread.thread_id)["token_counts"] += 1
        responses.append(
            schema.row(
                "responses",
                thread_id=thread.thread_id,
                turn_id=response["turn_id"],
                root_turn_id=response["turn_id"],
                session_id=response.get("producer") or thread.parent or thread.thread_id,
                response_id=response["response_id"],
                ts=response["ts"],
                rollout_ordinal=response["rollout_ordinal"],
                model=response.get("model"),
                input_tokens=total_input,
                cached_input_tokens=cached,
                cache_write_input_tokens=cache_write,
                cache_write_1h_input_tokens=cache_write_1h,
                output_tokens=output,
                reasoning_output_tokens=None,
                wasted_output_tokens=wasted,
                total_tokens=total,
                turn_input_tokens=turn["input"],
                turn_cached_input_tokens=turn["cached"],
                turn_output_tokens=turn["output"],
                turn_total_tokens=turn["total"],
                thread_input_tokens=thread_totals["input"],
                thread_cached_input_tokens=thread_totals["cached"],
                thread_output_tokens=thread_totals["output"],
                thread_total_tokens=thread_totals["total"],
                source=SOURCE,
                origin=COPIED if copied else "assistant",
            )
        )
        if copied:
            continue
        counts.append(
            schema.row(
                "token_counts",
                thread_id=thread.thread_id,
                turn_id=response["turn_id"],
                ts=response["ts"],
                rollout_ordinal=response["rollout_ordinal"],
                total_input_tokens=thread_totals["input"],
                total_cached_input_tokens=thread_totals["cached"],
                total_cache_write_input_tokens=thread_totals["cache_write"],
                total_output_tokens=thread_totals["output"],
                total_tokens=thread_totals["total"],
                last_input_tokens=total_input,
                last_cached_input_tokens=cached,
                last_cache_write_input_tokens=cache_write,
                last_output_tokens=output,
                last_total_tokens=total,
                source=SOURCE,
                origin="assistant",
            )
        )
    return responses, counts

default_claude_dir()

Return $CLAUDE_CONFIG_DIR or ~/.claude.

Source code in packages/agent-history/src/pdum/agent_history/claude.py

def default_claude_dir() -> Path:
    """Return ``$CLAUDE_CONFIG_DIR`` or ``~/.claude``."""
    return Path(os.environ.get("CLAUDE_CONFIG_DIR") or Path.home() / ".claude").expanduser()

ingest_claude_dir(claude_dir=None, db_path=None, *, window_days=DEFAULT_WINDOW_DAYS, blobs=True, now=None, mode='rebuild')

Ingest a Claude Code configuration directory into a DuckDB file.

Parameters:

Name	Type	Description	Default
claude_dir	Path or None	Configuration directory; defaults to $CLAUDE_CONFIG_DIR or ~/.claude.	None
db_path	Target or Path or str or None	Target DuckDB file or md:<database>; defaults to claude-history.duckdb in the current directory.	None
window_days	int or None	Support window in days; None ingests everything.	DEFAULT_WINDOW_DAYS
blobs	bool	Store tool calls with results and message blocks in item_blobs.	True
now	datetime or None	Reference time for the window, for tests.	None
mode	str	rebuild (fresh local file), sync (incremental) or full (sync re-reading everything).	'rebuild'

Returns:

Type	Description
IngestSummary	Phase timings, row counts and notes.

Source code in packages/agent-history/src/pdum/agent_history/claude.py

def ingest_claude_dir(
    claude_dir: Path | None = None,
    db_path: Target | Path | str | None = None,
    *,
    window_days: int | None = DEFAULT_WINDOW_DAYS,
    blobs: bool = True,
    now: datetime | None = None,
    mode: str = "rebuild",
) -> IngestSummary:
    """Ingest a Claude Code configuration directory into a DuckDB file.

    Parameters
    ----------
    claude_dir : Path or None
        Configuration directory; defaults to ``$CLAUDE_CONFIG_DIR`` or ``~/.claude``.
    db_path : Target or Path or str or None
        Target DuckDB file or ``md:<database>``; defaults to ``claude-history.duckdb`` in the
        current directory.
    window_days : int or None
        Support window in days; None ingests everything.
    blobs : bool
        Store tool calls with results and message blocks in ``item_blobs``.
    now : datetime or None
        Reference time for the window, for tests.
    mode : str
        ``rebuild`` (fresh local file), ``sync`` (incremental) or ``full`` (sync re-reading everything).

    Returns
    -------
    IngestSummary
        Phase timings, row counts and notes.
    """
    ingest = ClaudeIngest(
        claude_dir or default_claude_dir(),
        db_path or Path("claude-history.duckdb"),
        window_days=window_days,
        blobs=blobs,
        now=now,
        mode=mode,
    )
    return ingest.run()

patch_line_counts(patch)

Count added and deleted lines in a Claude structuredPatch list of hunks.

Source code in packages/agent-history/src/pdum/agent_history/claude.py

def patch_line_counts(patch: Any) -> tuple[int, int] | None:
    """Count added and deleted lines in a Claude ``structuredPatch`` list of hunks."""
    if not isinstance(patch, list):
        return None
    added = deleted = 0
    for hunk in patch:
        for line in (hunk or {}).get("lines") or []:
            if isinstance(line, str):
                if line.startswith("+"):
                    added += 1
                elif line.startswith("-"):
                    deleted += 1
    return added, deleted

Scaffolding shared by the ingesters: provenance, run summary, staging, publishing, compaction.

A run scans its source for units (one transcript file each), asks the target's ledger which units changed, parses only those into a staging database, and publishes them into the target by replacing each changed thread's rows inside one transaction. The target is a local DuckDB file or a MotherDuck database; the same path serves both. Lineage is then recomputed from the store, see :mod:pdum.agent_history.derive.

CHILD_KEYED_TABLES = ('spawn_edges', 'fork_edges') module-attribute

Tables keyed by the child thread; the child's rows are replaced with it.

DEFAULT_WINDOW_DAYS = 183 module-attribute

Roughly six months; older history is counted, not modelled.

THREAD_KEYED_TABLES = ('threads', 'turns', 'responses', 'token_counts', 'items', 'item_blobs', 'rollout_events', 'response_items', 'coverage', 'thread_links') module-attribute

Tables whose rows belong to one thread and are replaced with it.

BaseIngest

Common run scaffolding for one source into one target.

Parameters:

Name	Type	Description	Default
source	str	The source value written to every row, such as codex-fs.	required
agent	str	Agent kind recorded in provenance, codex or claude.	required
root	Path	Directory to read.	required
target	Target or Path or str	Where to publish: a local DuckDB file or md:<database>.	required
window_days	int or None	Threads anchored before now - window_days are counted, not modelled. None disables.	DEFAULT_WINDOW_DAYS
blobs	bool	Store full item JSON in item_blobs.	True
now	datetime or None	Reference time for the window; defaults to the current UTC time.	None
host	dict or None	Host identity; defaults to :func:host_identity.	None
mode	str	rebuild replaces a local file outright (the default of pdum-history ingest); sync replaces only the threads whose files changed since the ledger was written; full is a sync that re-reads every unit, used after a schema change.	'rebuild'

Source code in packages/agent-history/src/pdum/agent_history/base.py

class BaseIngest:
    """Common run scaffolding for one source into one target.

    Parameters
    ----------
    source : str
        The ``source`` value written to every row, such as ``codex-fs``.
    agent : str
        Agent kind recorded in provenance, ``codex`` or ``claude``.
    root : Path
        Directory to read.
    target : Target or Path or str
        Where to publish: a local DuckDB file or ``md:<database>``.
    window_days : int or None
        Threads anchored before ``now - window_days`` are counted, not modelled. None disables.
    blobs : bool
        Store full item JSON in ``item_blobs``.
    now : datetime or None
        Reference time for the window; defaults to the current UTC time.
    host : dict or None
        Host identity; defaults to :func:`host_identity`.
    mode : str
        ``rebuild`` replaces a local file outright (the default of ``pdum-history ingest``);
        ``sync`` replaces only the threads whose files changed since the ledger was written;
        ``full`` is a sync that re-reads every unit, used after a schema change.
    """

    def __init__(
        self,
        source: str,
        agent: str,
        root: Path,
        target: Target | Path | str,
        *,
        window_days: int | None = DEFAULT_WINDOW_DAYS,
        blobs: bool = True,
        now: datetime | None = None,
        host: dict[str, str] | None = None,
        mode: str = "rebuild",
    ) -> None:
        if mode not in {"rebuild", "sync", "full"}:
            raise ValueError(f"unknown mode {mode!r}")
        self.source = source
        self.agent = agent
        self.root = root.expanduser().resolve()
        self.target = target if isinstance(target, Target) else Target(str(target))
        if mode == "rebuild" and self.target.motherduck:
            mode = "full"
        self.mode = mode
        self.blobs = blobs
        reference = now or datetime.now(UTC).replace(tzinfo=None)
        self.cutoff = reference - timedelta(days=window_days) if window_days is not None else None
        self.run_id = uuid.uuid4().hex[:12]
        self.host = host or host_identity()
        root_hash = hashlib.sha1(str(self.root).encode()).hexdigest()[:8]
        self.source_id = f"{agent}-{self.host['host_id'][:8]}-{root_hash}"
        self.versions: set[str] = set()
        self.summary = IngestSummary(
            run_id=self.run_id,
            source=source,
            source_id=self.source_id,
            root=str(self.root),
            target=self.target.spec,
            mode=mode,
            window_cutoff=self.cutoff.isoformat() if self.cutoff else None,
            host={k: v for k, v in self.host.items() if k != "created_at"},
        )
        self.coverage: dict[str, dict[str, Any]] = {}
        self.unit_threads: dict[str, list[str]] = {}
        """Thread ids each unit produced this run, for the ledger and for later replacement."""
        self.con: duckdb.DuckDBPyConnection | None = None
        self._source_index = {
            table: [index for index, (name, _) in enumerate(columns) if name == "source_id"]
            for table, columns in schema.COLUMNS.items()
        }

    # ---------------------------------------------------------------- helpers
    @property
    def db_path(self) -> Path:
        """The local target file; raises for MotherDuck targets."""
        return self.target.path

    def in_window(self, moment: datetime | None) -> bool:
        """Return whether a thread anchored at ``moment`` is modelled."""
        if self.cutoff is None:
            return True
        return moment is not None and moment >= self.cutoff

    def connection(self) -> duckdb.DuckDBPyConnection:
        assert self.con is not None, "run() opens the database"
        return self.con

    def write(self, table: str, rows: list[tuple]) -> int:
        """Insert rows, stamping this run's ``source_id`` where the row left it empty."""
        positions = self._source_index.get(table) or []
        if positions and rows:
            index = positions[0]
            rows = [
                row if row[index] is not None else row[:index] + (self.source_id,) + row[index + 1 :] for row in rows
            ]
        return schema.insert(self.connection(), table, rows)

    def phase(self, name: str, started: float, rows: int, detail: str = "") -> None:
        """Record one phase's timing in the summary and in ``ingest_runs``."""
        seconds = time.perf_counter() - started
        self.summary.phases[name] = {"seconds": round(seconds, 3), "rows": rows, "detail": detail}
        self.write(
            "ingest_runs",
            [
                schema.row(
                    "ingest_runs",
                    run_id=self.run_id,
                    source=self.source,
                    phase=name,
                    started_at=datetime.now(UTC).replace(tzinfo=None),
                    seconds=seconds,
                    rows=rows,
                    detail=detail,
                )
            ],
        )

    def cover(self, thread_id: str) -> dict[str, Any]:
        """Return the mutable coverage record for a thread, creating it on first use."""
        return self.coverage.setdefault(
            thread_id,
            {
                "thread_id": thread_id,
                "in_window": True,
                "in_catalogue": False,
                "in_projection": False,
                "rollout_format": None,
                "rollout_lines": 0,
                "turn_contexts": 0,
                "usage_records": 0,
                "token_counts": 0,
                "projection_turns": 0,
                "projection_items": 0,
                "rollout_item_completed": 0,
                "response_items": 0,
                "compactions": 0,
                "unknown_line_types": 0,
                "marker_lines": 0,
            },
        )

    def unknown(self, kind: str) -> None:
        self.summary.unknown_line_types[kind] = self.summary.unknown_line_types.get(kind, 0) + 1

    def ignored(self, kind: str) -> None:
        self.summary.ignored_line_types[kind] = self.summary.ignored_line_types.get(kind, 0) + 1

    def write_coverage_and_meta(self, extra_meta: dict[str, str] | None = None) -> int:
        """Write ``coverage``, ``provenance`` and ``meta`` rows; return the number of coverage rows."""
        cover_rows = [
            schema.row("coverage", **cover, source=self.source, origin=f"ingest-{self.source}")
            for cover in self.coverage.values()
        ]
        self.write("coverage", cover_rows)
        now = datetime.now(UTC).replace(tzinfo=None)
        self.write(
            "provenance",
            [
                schema.row(
                    "provenance",
                    source_id=self.source_id,
                    host_id=self.host["host_id"],
                    hostname=self.host.get("hostname"),
                    os_user=getpass.getuser(),
                    platform=self.host.get("platform"),
                    agent=self.agent,
                    root=str(self.root),
                    agent_versions=json.dumps(sorted(self.versions)),
                    window_cutoff=self.cutoff,
                    schema_version=schema.SCHEMA_VERSION,
                    ingester_version=ingester_version(),
                    ingested_at=now,
                    run_id=self.run_id,
                )
            ],
        )
        meta = {
            "schema_version": str(schema.SCHEMA_VERSION),
            "last_run_id": self.run_id,
            "last_ingested_at": now.isoformat(),
            **(extra_meta or {}),
        }
        self.write("meta", [schema.row("meta", key=key, value=value) for key, value in meta.items()])
        return len(cover_rows)

    # -------------------------------------------------------------------- run
    def run(self) -> IngestSummary:
        """Scan, plan, ingest the changed units, publish, derive lineage and return the summary."""
        total = time.perf_counter()
        scan = self.scan()
        if self.mode == "rebuild":
            self._rebuild(scan)
        else:
            self._sync(scan)
        self.summary.total_seconds = round(time.perf_counter() - total, 3)
        return self.summary

    def _rebuild(self, scan: dict[str, ledger_module.Unit]) -> None:
        """Write a fresh local file directly, then compact it."""
        from .derive import dedupe_responses, derive_lineage

        path = self.db_path
        if path.exists():
            path.unlink()
        path.parent.mkdir(parents=True, exist_ok=True)
        plan = ledger_module.plan(scan, {}, full=True)
        self.summary.units = plan.counts
        self.con = duckdb.connect(str(path))
        try:
            schema.create(self.con)
            self.ingest(plan.changed)
            now = datetime.now(UTC).replace(tzinfo=None)
            self.write(
                "ingest_ledger", ledger_module.rows(self.source_id, self.run_id, plan, {}, self.unit_threads, now)
            )
            self.summary.derived = derive_lineage(self.con, self.source_id, self.source)
            self.summary.derived["deduplicated"] = dedupe_responses(self.con, self.source_id)
            self.summary.notes.extend(self.summary.derived.get("notes", []))
            self.summary.tables = {
                table: self.con.execute(f"select count(*) from {table}").fetchone()[0] for table in schema.COLUMNS
            }
            self.con.execute("CHECKPOINT")
        finally:
            self.con.close()
            self.con = None
        compact(path)
        self.summary.db_bytes = path.stat().st_size

    def _sync(self, scan: dict[str, ledger_module.Unit]) -> None:
        """Parse changed units into a staging file and publish them into the target."""
        target_con = self.target.connect()
        try:
            ensure_schema(target_con, rebuild=self.mode == "full")
            stored = ledger_module.read(target_con, self.source_id)
            self.prepare(target_con)
        finally:
            target_con.close()
        plan = ledger_module.plan(scan, stored, full=self.mode == "full")
        self.summary.units = plan.counts
        staging_dir = Path(tempfile.mkdtemp(prefix="pdum-history-"))
        staging = staging_dir / "staging.duckdb"
        self.con = duckdb.connect(str(staging))
        try:
            schema.create(self.con)
            self.ingest(plan.changed)
            now = datetime.now(UTC).replace(tzinfo=None)
            self.write(
                "ingest_ledger", ledger_module.rows(self.source_id, self.run_id, plan, stored, self.unit_threads, now)
            )
            self.summary.tables = {
                table: self.con.execute(f"select count(*) from {table}").fetchone()[0] for table in schema.COLUMNS
            }
            self.con.execute("CHECKPOINT")
        finally:
            self.con.close()
            self.con = None
        try:
            self.publish(staging, plan, stored)
        finally:
            staging.unlink(missing_ok=True)
            staging.with_name(staging.name + ".wal").unlink(missing_ok=True)
            try:
                staging_dir.rmdir()
            except OSError:  # pragma: no cover - leftover temp files
                pass

    def publish(self, staging: Path, plan: ledger_module.Plan, stored: dict[str, ledger_module.LedgerRow]) -> None:
        """Replace the changed threads' rows in the target with the staged rows, then derive lineage."""
        from .derive import dedupe_responses, derive_lineage

        started = time.perf_counter()
        replaced: set[str] = set()
        for unit_id in plan.changed:
            replaced.update(self.unit_threads.get(unit_id) or [unit_id])
            row = stored.get(unit_id)
            if row is not None:
                replaced.update(row.thread_ids)
        con = self.target.connect()
        alias = f"staging_{self.run_id}"
        try:
            # MotherDuck remembers attachments across sessions, so never reuse a name and always detach.
            con.execute(f"DETACH DATABASE IF EXISTS {alias}")
            con.execute(f"ATTACH '{staging}' AS {alias} (READ_ONLY)")
            con.execute("BEGIN TRANSACTION")
            con.execute("CREATE TEMP TABLE _replaced (thread_id VARCHAR)")
            if replaced:
                con.executemany("INSERT INTO _replaced VALUES (?)", [(thread_id,) for thread_id in sorted(replaced)])
            for table in THREAD_KEYED_TABLES:
                con.execute(
                    f"DELETE FROM {table} WHERE source_id = ? AND thread_id IN (SELECT thread_id FROM _replaced)",
                    [self.source_id],
                )
            for table in CHILD_KEYED_TABLES:
                con.execute(
                    f"DELETE FROM {table} WHERE source_id = ? AND child_thread_id IN (SELECT thread_id FROM _replaced)",
                    [self.source_id],
                )
            if self.mode == "full":
                con.execute("DELETE FROM logs WHERE source_id = ?", [self.source_id])
            con.execute("DELETE FROM provenance WHERE source_id = ?", [self.source_id])
            con.execute(f"DELETE FROM meta WHERE key IN (SELECT key FROM {alias}.meta)")
            touched_units = sorted({*plan.changed, *plan.gone})
            if touched_units:
                con.execute("CREATE TEMP TABLE _units (unit_id VARCHAR)")
                con.executemany("INSERT INTO _units VALUES (?)", [(unit_id,) for unit_id in touched_units])
                con.execute(
                    "DELETE FROM ingest_ledger WHERE source_id = ? AND unit_id IN (SELECT unit_id FROM _units)",
                    [self.source_id],
                )
            for table in schema.COLUMNS:
                con.execute(f"INSERT INTO {table} SELECT * FROM {alias}.{table}")
            deduplicated = dedupe_responses(con, self.source_id)
            con.execute("COMMIT")
            self.summary.derived = derive_lineage(con, self.source_id, self.source)
            self.summary.derived["deduplicated"] = deduplicated
            self.summary.notes.extend(self.summary.derived.get("notes", []))
            if not self.target.motherduck:
                con.execute("CHECKPOINT")
                self.summary.db_bytes = self.db_path.stat().st_size
        except Exception:
            try:
                con.execute("ROLLBACK")
            except duckdb.Error:  # pragma: no cover - nothing to roll back
                pass
            raise
        finally:
            try:
                con.execute(f"DETACH DATABASE IF EXISTS {alias}")
            except duckdb.Error:  # pragma: no cover - connection already unusable
                pass
            con.close()
        self.summary.phases["publish"] = {
            "seconds": round(time.perf_counter() - started, 3),
            "rows": sum(self.summary.tables.values()),
            "detail": f"threads replaced {len(replaced)}; units gone {len(plan.gone)}",
        }

    # ------------------------------------------------------------- subclasses
    def scan(self) -> dict[str, ledger_module.Unit]:  # pragma: no cover - subclasses implement
        """Return every unit the source holds, keyed by unit id, from file metadata alone."""
        raise NotImplementedError

    def prepare(self, target: duckdb.DuckDBPyConnection) -> None:
        """Read anything a sync needs from the target before parsing, such as a log watermark."""

    def ingest(self, units: dict[str, ledger_module.Unit]) -> None:  # pragma: no cover - subclasses implement
        """Parse the given units and write their rows."""
        raise NotImplementedError

db_path property

The local target file; raises for MotherDuck targets.

unit_threads = {} instance-attribute

Thread ids each unit produced this run, for the ledger and for later replacement.

cover(thread_id)

Return the mutable coverage record for a thread, creating it on first use.

Source code in packages/agent-history/src/pdum/agent_history/base.py

def cover(self, thread_id: str) -> dict[str, Any]:
    """Return the mutable coverage record for a thread, creating it on first use."""
    return self.coverage.setdefault(
        thread_id,
        {
            "thread_id": thread_id,
            "in_window": True,
            "in_catalogue": False,
            "in_projection": False,
            "rollout_format": None,
            "rollout_lines": 0,
            "turn_contexts": 0,
            "usage_records": 0,
            "token_counts": 0,
            "projection_turns": 0,
            "projection_items": 0,
            "rollout_item_completed": 0,
            "response_items": 0,
            "compactions": 0,
            "unknown_line_types": 0,
            "marker_lines": 0,
        },
    )

in_window(moment)

Return whether a thread anchored at moment is modelled.

Source code in packages/agent-history/src/pdum/agent_history/base.py

def in_window(self, moment: datetime | None) -> bool:
    """Return whether a thread anchored at ``moment`` is modelled."""
    if self.cutoff is None:
        return True
    return moment is not None and moment >= self.cutoff

ingest(units)

Parse the given units and write their rows.

Source code in packages/agent-history/src/pdum/agent_history/base.py

def ingest(self, units: dict[str, ledger_module.Unit]) -> None:  # pragma: no cover - subclasses implement
    """Parse the given units and write their rows."""
    raise NotImplementedError

phase(name, started, rows, detail='')

Record one phase's timing in the summary and in ingest_runs.

Source code in packages/agent-history/src/pdum/agent_history/base.py

def phase(self, name: str, started: float, rows: int, detail: str = "") -> None:
    """Record one phase's timing in the summary and in ``ingest_runs``."""
    seconds = time.perf_counter() - started
    self.summary.phases[name] = {"seconds": round(seconds, 3), "rows": rows, "detail": detail}
    self.write(
        "ingest_runs",
        [
            schema.row(
                "ingest_runs",
                run_id=self.run_id,
                source=self.source,
                phase=name,
                started_at=datetime.now(UTC).replace(tzinfo=None),
                seconds=seconds,
                rows=rows,
                detail=detail,
            )
        ],
    )

prepare(target)

Read anything a sync needs from the target before parsing, such as a log watermark.

Source code in packages/agent-history/src/pdum/agent_history/base.py

def prepare(self, target: duckdb.DuckDBPyConnection) -> None:
    """Read anything a sync needs from the target before parsing, such as a log watermark."""

publish(staging, plan, stored)

Replace the changed threads' rows in the target with the staged rows, then derive lineage.

Source code in packages/agent-history/src/pdum/agent_history/base.py

def publish(self, staging: Path, plan: ledger_module.Plan, stored: dict[str, ledger_module.LedgerRow]) -> None:
    """Replace the changed threads' rows in the target with the staged rows, then derive lineage."""
    from .derive import dedupe_responses, derive_lineage

    started = time.perf_counter()
    replaced: set[str] = set()
    for unit_id in plan.changed:
        replaced.update(self.unit_threads.get(unit_id) or [unit_id])
        row = stored.get(unit_id)
        if row is not None:
            replaced.update(row.thread_ids)
    con = self.target.connect()
    alias = f"staging_{self.run_id}"
    try:
        # MotherDuck remembers attachments across sessions, so never reuse a name and always detach.
        con.execute(f"DETACH DATABASE IF EXISTS {alias}")
        con.execute(f"ATTACH '{staging}' AS {alias} (READ_ONLY)")
        con.execute("BEGIN TRANSACTION")
        con.execute("CREATE TEMP TABLE _replaced (thread_id VARCHAR)")
        if replaced:
            con.executemany("INSERT INTO _replaced VALUES (?)", [(thread_id,) for thread_id in sorted(replaced)])
        for table in THREAD_KEYED_TABLES:
            con.execute(
                f"DELETE FROM {table} WHERE source_id = ? AND thread_id IN (SELECT thread_id FROM _replaced)",
                [self.source_id],
            )
        for table in CHILD_KEYED_TABLES:
            con.execute(
                f"DELETE FROM {table} WHERE source_id = ? AND child_thread_id IN (SELECT thread_id FROM _replaced)",
                [self.source_id],
            )
        if self.mode == "full":
            con.execute("DELETE FROM logs WHERE source_id = ?", [self.source_id])
        con.execute("DELETE FROM provenance WHERE source_id = ?", [self.source_id])
        con.execute(f"DELETE FROM meta WHERE key IN (SELECT key FROM {alias}.meta)")
        touched_units = sorted({*plan.changed, *plan.gone})
        if touched_units:
            con.execute("CREATE TEMP TABLE _units (unit_id VARCHAR)")
            con.executemany("INSERT INTO _units VALUES (?)", [(unit_id,) for unit_id in touched_units])
            con.execute(
                "DELETE FROM ingest_ledger WHERE source_id = ? AND unit_id IN (SELECT unit_id FROM _units)",
                [self.source_id],
            )
        for table in schema.COLUMNS:
            con.execute(f"INSERT INTO {table} SELECT * FROM {alias}.{table}")
        deduplicated = dedupe_responses(con, self.source_id)
        con.execute("COMMIT")
        self.summary.derived = derive_lineage(con, self.source_id, self.source)
        self.summary.derived["deduplicated"] = deduplicated
        self.summary.notes.extend(self.summary.derived.get("notes", []))
        if not self.target.motherduck:
            con.execute("CHECKPOINT")
            self.summary.db_bytes = self.db_path.stat().st_size
    except Exception:
        try:
            con.execute("ROLLBACK")
        except duckdb.Error:  # pragma: no cover - nothing to roll back
            pass
        raise
    finally:
        try:
            con.execute(f"DETACH DATABASE IF EXISTS {alias}")
        except duckdb.Error:  # pragma: no cover - connection already unusable
            pass
        con.close()
    self.summary.phases["publish"] = {
        "seconds": round(time.perf_counter() - started, 3),
        "rows": sum(self.summary.tables.values()),
        "detail": f"threads replaced {len(replaced)}; units gone {len(plan.gone)}",
    }

run()

Scan, plan, ingest the changed units, publish, derive lineage and return the summary.

Source code in packages/agent-history/src/pdum/agent_history/base.py

def run(self) -> IngestSummary:
    """Scan, plan, ingest the changed units, publish, derive lineage and return the summary."""
    total = time.perf_counter()
    scan = self.scan()
    if self.mode == "rebuild":
        self._rebuild(scan)
    else:
        self._sync(scan)
    self.summary.total_seconds = round(time.perf_counter() - total, 3)
    return self.summary

scan()

Return every unit the source holds, keyed by unit id, from file metadata alone.

Source code in packages/agent-history/src/pdum/agent_history/base.py

def scan(self) -> dict[str, ledger_module.Unit]:  # pragma: no cover - subclasses implement
    """Return every unit the source holds, keyed by unit id, from file metadata alone."""
    raise NotImplementedError

write(table, rows)

Insert rows, stamping this run's source_id where the row left it empty.

Source code in packages/agent-history/src/pdum/agent_history/base.py

def write(self, table: str, rows: list[tuple]) -> int:
    """Insert rows, stamping this run's ``source_id`` where the row left it empty."""
    positions = self._source_index.get(table) or []
    if positions and rows:
        index = positions[0]
        rows = [
            row if row[index] is not None else row[:index] + (self.source_id,) + row[index + 1 :] for row in rows
        ]
    return schema.insert(self.connection(), table, rows)

write_coverage_and_meta(extra_meta=None)

Write coverage, provenance and meta rows; return the number of coverage rows.

Source code in packages/agent-history/src/pdum/agent_history/base.py

def write_coverage_and_meta(self, extra_meta: dict[str, str] | None = None) -> int:
    """Write ``coverage``, ``provenance`` and ``meta`` rows; return the number of coverage rows."""
    cover_rows = [
        schema.row("coverage", **cover, source=self.source, origin=f"ingest-{self.source}")
        for cover in self.coverage.values()
    ]
    self.write("coverage", cover_rows)
    now = datetime.now(UTC).replace(tzinfo=None)
    self.write(
        "provenance",
        [
            schema.row(
                "provenance",
                source_id=self.source_id,
                host_id=self.host["host_id"],
                hostname=self.host.get("hostname"),
                os_user=getpass.getuser(),
                platform=self.host.get("platform"),
                agent=self.agent,
                root=str(self.root),
                agent_versions=json.dumps(sorted(self.versions)),
                window_cutoff=self.cutoff,
                schema_version=schema.SCHEMA_VERSION,
                ingester_version=ingester_version(),
                ingested_at=now,
                run_id=self.run_id,
            )
        ],
    )
    meta = {
        "schema_version": str(schema.SCHEMA_VERSION),
        "last_run_id": self.run_id,
        "last_ingested_at": now.isoformat(),
        **(extra_meta or {}),
    }
    self.write("meta", [schema.row("meta", key=key, value=value) for key, value in meta.items()])
    return len(cover_rows)

IngestSummary dataclass

What one ingest run read and wrote.

Parameters:

Name	Type	Description	Default
run_id	str	Identifier shared by the run's ingest_runs rows.	required
source	str	The source value written to every row.	required
source_id	str	Provenance key written to every row; stable for one agent, machine and root.	required
root	str	Directory that was read.	required
target	str	Where rows were published: a file path or md:<database>.	required
mode	str	rebuild for a fresh local file, sync for an incremental run, full for a sync that re-read every unit.	required
window_cutoff	str or None	ISO timestamp before which threads are only counted, or None when unlimited.	required

Source code in packages/agent-history/src/pdum/agent_history/base.py

@dataclass
class IngestSummary:
    """What one ingest run read and wrote.

    Parameters
    ----------
    run_id : str
        Identifier shared by the run's ``ingest_runs`` rows.
    source : str
        The ``source`` value written to every row.
    source_id : str
        Provenance key written to every row; stable for one agent, machine and root.
    root : str
        Directory that was read.
    target : str
        Where rows were published: a file path or ``md:<database>``.
    mode : str
        ``rebuild`` for a fresh local file, ``sync`` for an incremental run, ``full`` for a sync
        that re-read every unit.
    window_cutoff : str or None
        ISO timestamp before which threads are only counted, or None when unlimited.
    """

    run_id: str
    source: str
    source_id: str
    root: str
    target: str
    mode: str
    window_cutoff: str | None
    label: str = ""
    host: dict[str, str] = field(default_factory=dict)
    units: dict[str, int] = field(default_factory=dict)
    phases: dict[str, dict[str, Any]] = field(default_factory=dict)
    tables: dict[str, int] = field(default_factory=dict)
    derived: dict[str, Any] = field(default_factory=dict)
    unknown_line_types: dict[str, int] = field(default_factory=dict)
    ignored_line_types: dict[str, int] = field(default_factory=dict)
    notes: list[str] = field(default_factory=list)
    db_bytes: int = 0
    total_seconds: float = 0.0

    @property
    def db_path(self) -> str:
        """The target spec, kept under its old name for callers that printed it."""
        return self.target

    def to_dict(self) -> dict[str, Any]:
        """Return a JSON-compatible dictionary."""
        return asdict(self)

db_path property

The target spec, kept under its old name for callers that printed it.

to_dict()

Return a JSON-compatible dictionary.

Source code in packages/agent-history/src/pdum/agent_history/base.py

def to_dict(self) -> dict[str, Any]:
    """Return a JSON-compatible dictionary."""
    return asdict(self)

assign_lineage(threads)

Fill lineage_id and lineage_depth on every thread record from its predecessor chain.

A thread's predecessor is forked_from_id (the session it continued or forked from) or, for a sub-agent, parent_thread_id. The root of the chain is the lineage id and the hop count is the depth. A chain that leaves the corpus ends at the missing predecessor, which still names the lineage so siblings of a deleted session stay together; a cycle breaks at its first repeat.

Parameters:

Name	Type	Description	Default
threads	dict	Thread records by id, updated in place.	required

Returns:

Type	Description
dict	{thread_id: reason} for chains that cycled.

Source code in packages/agent-history/src/pdum/agent_history/base.py

def assign_lineage(threads: dict[str, dict[str, Any]]) -> dict[str, str]:
    """Fill ``lineage_id`` and ``lineage_depth`` on every thread record from its predecessor chain.

    A thread's predecessor is ``forked_from_id`` (the session it continued or forked from) or, for
    a sub-agent, ``parent_thread_id``. The root of the chain is the lineage id and the hop count is
    the depth. A chain that leaves the corpus ends at the missing predecessor, which still names the
    lineage so siblings of a deleted session stay together; a cycle breaks at its first repeat.

    Parameters
    ----------
    threads : dict
        Thread records by id, updated in place.

    Returns
    -------
    dict
        ``{thread_id: reason}`` for chains that cycled.
    """
    notes: dict[str, str] = {}
    memo: dict[str, tuple[str, int]] = {}

    def resolve(thread_id: str, seen: tuple[str, ...]) -> tuple[str, int]:
        if thread_id in memo:
            return memo[thread_id]
        record = threads.get(thread_id)
        if record is None:
            return thread_id, 0
        predecessor = record.get("forked_from_id") or record.get("parent_thread_id")
        if not predecessor or predecessor == thread_id:
            result = (thread_id, 0)
        elif predecessor in seen:
            notes[thread_id] = f"lineage of {thread_id} cycles through {predecessor}"
            result = (thread_id, 0)
        else:
            root, depth = resolve(predecessor, (*seen, thread_id))
            result = (root, depth + 1)
        memo[thread_id] = result
        return result

    for thread_id, record in threads.items():
        root, depth = resolve(thread_id, ())
        record["lineage_id"] = root
        record["lineage_depth"] = depth
    return notes

compact(db_path)

Rewrite a freshly written DuckDB file without the free blocks bulk inserts leave behind.

Parameters:

Name	Type	Description	Default
db_path	Path	Database to compact in place; a sibling temporary file is written and renamed over it.	required

Source code in packages/agent-history/src/pdum/agent_history/base.py

def compact(db_path: Path) -> None:
    """Rewrite a freshly written DuckDB file without the free blocks bulk inserts leave behind.

    Parameters
    ----------
    db_path : Path
        Database to compact in place; a sibling temporary file is written and renamed over it.
    """
    temporary = db_path.with_name(f".{db_path.name}.compact")
    temporary.unlink(missing_ok=True)
    connection = duckdb.connect(str(temporary))
    try:
        target = connection.execute("select current_database()").fetchone()[0]
        connection.execute(f"ATTACH '{db_path}' AS source (READ_ONLY)")
        connection.execute(f'COPY FROM DATABASE source TO "{target}"')
        connection.execute("CHECKPOINT")
    finally:
        connection.close()
    os.replace(temporary, db_path)

host_identity(state_dir=None)

Return this machine's stable identity, minting host_id on first use.

A hostname is a label, never a key: machines get renamed and a rename would fork one machine's history into two. The id is minted once and kept under the launcher's own data directory, never inside an agent's home, and the hostname rides along as a label that may change. PDUM_AGENT_HISTORY_HOST_ID overrides the stored id, for tests and containers.

Parameters:

Name	Type	Description	Default
state_dir	Path or None	Directory holding host.json; defaults to the platform user data directory.	None

Returns:

Type	Description
dict	host_id, hostname, platform and created_at.

Source code in packages/agent-history/src/pdum/agent_history/base.py

def host_identity(state_dir: Path | None = None) -> dict[str, str]:
    """Return this machine's stable identity, minting ``host_id`` on first use.

    A hostname is a label, never a key: machines get renamed and a rename would fork one
    machine's history into two. The id is minted once and kept under the launcher's own data
    directory, never inside an agent's home, and the hostname rides along as a label that may
    change. ``PDUM_AGENT_HISTORY_HOST_ID`` overrides the stored id, for tests and containers.

    Parameters
    ----------
    state_dir : Path or None
        Directory holding ``host.json``; defaults to the platform user data directory.

    Returns
    -------
    dict
        ``host_id``, ``hostname``, ``platform`` and ``created_at``.
    """
    hostname = socket.gethostname()
    system = f"{platform.system().lower()} {platform.release()}"
    override = os.environ.get("PDUM_AGENT_HISTORY_HOST_ID")
    if override:
        return {"host_id": override, "hostname": hostname, "platform": system, "created_at": ""}
    directory = state_dir or platformdirs.user_data_path("pdum-agent-history", appauthor=False)
    path = directory / "host.json"
    stored: dict[str, str] = {}
    try:
        stored = json.loads(path.read_text())
    except (OSError, ValueError):
        stored = {}
    identity = {
        "host_id": stored.get("host_id") or uuid.uuid4().hex,
        "hostname": hostname,
        "platform": system,
        "created_at": stored.get("created_at") or datetime.now(UTC).isoformat(),
    }
    if identity != stored:
        try:
            directory.mkdir(parents=True, exist_ok=True)
            path.write_text(json.dumps(identity, indent=2) + "\n")
        except OSError:  # pragma: no cover - read-only home
            pass
    return identity

ingester_version()

Return the installed distribution version, or dev from a source checkout.

Source code in packages/agent-history/src/pdum/agent_history/base.py

def ingester_version() -> str:
    """Return the installed distribution version, or ``dev`` from a source checkout."""
    try:
        return package_version("habemus-papadum-agent-history")
    except PackageNotFoundError:  # pragma: no cover - depends on the environment
        return "dev"

repo_from_origin(origin)

Normalize a git remote URL to owner/name when it has that shape.

Parameters:

Name	Type	Description	Default
origin	str or None	https://host/owner/name.git, git@host:owner/name.git or similar.	required

Returns:

Type	Description
str or None	The last two path segments without .git; one segment when that is all there is.

Source code in packages/agent-history/src/pdum/agent_history/base.py

def repo_from_origin(origin: str | None) -> str | None:
    """Normalize a git remote URL to ``owner/name`` when it has that shape.

    Parameters
    ----------
    origin : str or None
        ``https://host/owner/name.git``, ``git@host:owner/name.git`` or similar.

    Returns
    -------
    str or None
        The last two path segments without ``.git``; one segment when that is all there is.
    """
    if not origin:
        return None
    text = origin.strip()
    if "://" in text:
        path = urlparse(text).path
    elif ":" in text and "@" in text.split(":", 1)[0]:
        path = text.split(":", 1)[1]
    else:
        path = text
    parts = [part for part in path.strip("/").split("/") if part]
    if not parts:
        return None
    parts[-1] = parts[-1].removesuffix(".git")
    return "/".join(parts[-2:])

One command for every agent on the machine: sync each configured source into each target.

build(source, target, config, *, mode, now=None, host=None)

Construct the ingester for one source and target.

Source code in packages/agent-history/src/pdum/agent_history/sync.py

def build(
    source: Source,
    target: Target,
    config: Config,
    *,
    mode: str,
    now: datetime | None = None,
    host: dict[str, str] | None = None,
) -> BaseIngest:
    """Construct the ingester for one source and target."""
    kind = CodexIngest if source.agent == "codex" else ClaudeIngest
    ingest = kind(
        source.root,
        target,
        window_days=config.window_days,
        blobs=config.blobs,
        now=now,
        mode=mode,
        host=host,
    )
    ingest.summary.label = source.label
    return ingest

sync(config, *, targets=None, full=False, now=None, host=None)

Sync every source into every selected target and return one summary per pair.

Parameters:

Name	Type	Description	Default
config	Config	Parsed configuration.	required
targets	iterable of str or None	Target names or specs to use; all when None.	None
full	bool	Re-read every unit instead of only the changed ones.	False
now	datetime or None	Reference time for the window, for tests.	None
host	dict or None	Host identity override, for tests.	None

Source code in packages/agent-history/src/pdum/agent_history/sync.py

def sync(
    config: Config,
    *,
    targets: Iterable[str] | None = None,
    full: bool = False,
    now: datetime | None = None,
    host: dict[str, str] | None = None,
) -> list[IngestSummary]:
    """Sync every source into every selected target and return one summary per pair.

    Parameters
    ----------
    config : Config
        Parsed configuration.
    targets : iterable of str or None
        Target names or specs to use; all when None.
    full : bool
        Re-read every unit instead of only the changed ones.
    now : datetime or None
        Reference time for the window, for tests.
    host : dict or None
        Host identity override, for tests.
    """
    wanted = set(targets or [])
    selected = [target for target in config.targets if not wanted or target.label in wanted or target.spec in wanted]
    if wanted and not selected:
        raise ConfigError(f"no target named {sorted(wanted)}; configured: {[t.label for t in config.targets]}")
    summaries: list[IngestSummary] = []
    for target in selected:
        for source in config.sources:
            ingest = build(source, target, config, mode="full" if full else "sync", now=now, host=host)
            summaries.append(ingest.run())
    return summaries

The per-machine description of what to ingest and where to publish it.

One YAML file lists sources (Codex homes, Claude directories, or every profile of a launcher) and targets (local DuckDB files, MotherDuck databases). pdum-history sync reads it so one command covers every agent on the machine. The default location is the user configuration directory, overridable with --config or PDUM_AGENT_HISTORY_CONFIG.

.. code-block:: yaml

window_days: 183
blobs: true
sources:
  - agent: codex
    home: ~/.codex
  - agent: codex
    profiles: all            # every Codex launcher profile on this machine
  - agent: claude
    dir: ~/.claude
  - agent: claude
    profiles: all            # every Claude launcher profile's claude/ directory
targets:
  - name: local
    duckdb: ~/history/agents.duckdb
  - name: cloud
    motherduck: agent_history   # md:agent_history, token from MOTHERDUCK_TOKEN
    token_env: MOTHERDUCK_BROWSER_TOKEN

Config dataclass

Parsed configuration.

Source code in packages/agent-history/src/pdum/agent_history/config.py

@dataclass
class Config:
    """Parsed configuration."""

    sources: list[Source]
    targets: list[Target]
    window_days: int | None = DEFAULT_WINDOW_DAYS
    blobs: bool = True
    path: Path | None = None
    notes: list[str] = field(default_factory=list)

ConfigError

Bases: ValueError

The configuration file is missing or malformed.

Source code in packages/agent-history/src/pdum/agent_history/config.py

class ConfigError(ValueError):
    """The configuration file is missing or malformed."""

Source dataclass

One directory to read.

Parameters:

Name	Type	Description	Default
agent	str	codex or claude.	required
root	Path	The Codex home or Claude configuration directory.	required
label	str	Where it came from, for output: home, dir or profile:<name>.	required

Source code in packages/agent-history/src/pdum/agent_history/config.py

@dataclass(frozen=True)
class Source:
    """One directory to read.

    Parameters
    ----------
    agent : str
        ``codex`` or ``claude``.
    root : Path
        The Codex home or Claude configuration directory.
    label : str
        Where it came from, for output: ``home``, ``dir`` or ``profile:<name>``.
    """

    agent: str
    root: Path
    label: str

default_config_path()

Return $PDUM_AGENT_HISTORY_CONFIG or history.yaml in the user configuration directory.

Source code in packages/agent-history/src/pdum/agent_history/config.py

def default_config_path() -> Path:
    """Return ``$PDUM_AGENT_HISTORY_CONFIG`` or ``history.yaml`` in the user configuration directory."""
    override = os.environ.get(CONFIG_ENV)
    if override:
        return Path(override).expanduser()
    return platformdirs.user_config_path("pdum-agent-history", appauthor=False) / "history.yaml"

load(path=None)

Read and validate the configuration file.

Parameters:

Name	Type	Description	Default
path	Path or None	File to read; defaults to :func:default_config_path.	None

Raises:

Type	Description
ConfigError	When the file is missing, not a mapping, or names an unknown agent or an empty target.

Source code in packages/agent-history/src/pdum/agent_history/config.py

def load(path: Path | None = None) -> Config:
    """Read and validate the configuration file.

    Parameters
    ----------
    path : Path or None
        File to read; defaults to :func:`default_config_path`.

    Raises
    ------
    ConfigError
        When the file is missing, not a mapping, or names an unknown agent or an empty target.
    """
    location = path or default_config_path()
    if not location.is_file():
        raise ConfigError(f"no configuration at {location}; write one (see `pdum-history sync --example`)")
    try:
        data = yaml.safe_load(location.read_text()) or {}
    except yaml.YAMLError as error:
        raise ConfigError(f"{location}: {error}") from error
    if not isinstance(data, dict):
        raise ConfigError(f"{location}: expected a mapping at the top level")
    config = Config(sources=[], targets=[], path=location)
    window = data.get("window_days", DEFAULT_WINDOW_DAYS)
    config.window_days = None if window in (None, 0, "all") else int(window)
    config.blobs = bool(data.get("blobs", True))
    for index, entry in enumerate(data.get("sources") or []):
        config.sources.extend(_sources(entry, index, config.notes))
    for index, entry in enumerate(data.get("targets") or []):
        config.targets.append(_target(entry, index))
    if not config.sources:
        raise ConfigError(f"{location}: no sources")
    if not config.targets:
        raise ConfigError(f"{location}: no targets")
    return config

profile_roots(agent, root)

List (name, directory) for every launcher profile under root.

A Codex profile directory is itself the Codex home. A Claude profile keeps its native configuration in a claude/ subdirectory.

Source code in packages/agent-history/src/pdum/agent_history/config.py

def profile_roots(agent: str, root: Path) -> list[tuple[str, Path]]:
    """List ``(name, directory)`` for every launcher profile under ``root``.

    A Codex profile directory is itself the Codex home. A Claude profile keeps its native
    configuration in a ``claude/`` subdirectory.
    """
    if not root.is_dir():
        return []
    found: list[tuple[str, Path]] = []
    for entry in sorted(root.iterdir()):
        if not entry.is_dir() or entry.name.startswith("."):
            continue
        directory = entry / "claude" if agent == "claude" else entry
        if directory.is_dir():
            found.append((entry.name, directory))
    return found

profiles_root(agent)

Return the launcher's default profiles directory for an agent.

Source code in packages/agent-history/src/pdum/agent_history/config.py

def profiles_root(agent: str) -> Path:
    """Return the launcher's default profiles directory for an agent."""
    return platformdirs.user_data_path(f"pdum-{agent}", appauthor=False) / "profiles"

Where a run's rows end up: a local DuckDB file or a MotherDuck database.

A target is named by a string: a filesystem path, or md:<database> for MotherDuck. The MotherDuck token comes from the environment (MOTHERDUCK_TOKEN, else motherduck_token, else the variable a target names), never from a file this package writes. Every run first writes into a staging database and then publishes into the target through one connection that sees both, so the same code path serves both kinds of target.

Target dataclass

One place rows are published to.

Parameters:

Name	Type	Description	Default
spec	str	A filesystem path or md:<database>.	required
name	str	Short label used in output; defaults to the spec.	''
token_env	str or None	Environment variable holding the MotherDuck token when it is not one of the defaults.	None

Source code in packages/agent-history/src/pdum/agent_history/targets.py

@dataclass(frozen=True)
class Target:
    """One place rows are published to.

    Parameters
    ----------
    spec : str
        A filesystem path or ``md:<database>``.
    name : str
        Short label used in output; defaults to the spec.
    token_env : str or None
        Environment variable holding the MotherDuck token when it is not one of the defaults.
    """

    spec: str
    name: str = ""
    token_env: str | None = None

    @property
    def label(self) -> str:
        return self.name or self.spec

    @property
    def motherduck(self) -> bool:
        return self.spec.startswith(MOTHERDUCK_PREFIX)

    @property
    def path(self) -> Path:
        """The local file, for filesystem targets."""
        if self.motherduck:
            raise TargetError(f"{self.spec} is not a local file")
        return Path(self.spec).expanduser().resolve()

    def token(self) -> str | None:
        """Return the MotherDuck token from the environment, or None when unset."""
        names = (*((self.token_env,) if self.token_env else ()), *TOKEN_VARIABLES)
        for name in names:
            value = os.environ.get(name)
            if value:
                return value
        return None

    def connect(self) -> duckdb.DuckDBPyConnection:
        """Open a read-write connection to the target, creating a local file when absent."""
        if self.motherduck:
            spec = self.spec if "?" in self.spec else f"{self.spec}?attach_mode=single"
            config = {"motherduck_token": self.token()} if self.token() else {}
            try:
                return duckdb.connect(spec, config=config)
            except duckdb.Error as error:
                raise TargetError(f"cannot connect to {self.spec}: {error}") from error
        self.path.parent.mkdir(parents=True, exist_ok=True)
        return duckdb.connect(str(self.path))

path property

The local file, for filesystem targets.

connect()

Open a read-write connection to the target, creating a local file when absent.

Source code in packages/agent-history/src/pdum/agent_history/targets.py

def connect(self) -> duckdb.DuckDBPyConnection:
    """Open a read-write connection to the target, creating a local file when absent."""
    if self.motherduck:
        spec = self.spec if "?" in self.spec else f"{self.spec}?attach_mode=single"
        config = {"motherduck_token": self.token()} if self.token() else {}
        try:
            return duckdb.connect(spec, config=config)
        except duckdb.Error as error:
            raise TargetError(f"cannot connect to {self.spec}: {error}") from error
    self.path.parent.mkdir(parents=True, exist_ok=True)
    return duckdb.connect(str(self.path))

token()

Return the MotherDuck token from the environment, or None when unset.

Source code in packages/agent-history/src/pdum/agent_history/targets.py

def token(self) -> str | None:
    """Return the MotherDuck token from the environment, or None when unset."""
    names = (*((self.token_env,) if self.token_env else ()), *TOKEN_VARIABLES)
    for name in names:
        value = os.environ.get(name)
        if value:
            return value
    return None

TargetError

Bases: RuntimeError

The target cannot be used as asked, for example because its schema version differs.

Source code in packages/agent-history/src/pdum/agent_history/targets.py

class TargetError(RuntimeError):
    """The target cannot be used as asked, for example because its schema version differs."""

ensure_schema(connection, *, rebuild)

Create the tables and views, or replace them when the version differs and a rebuild is allowed.

Parameters:

Name	Type	Description	Default
connection	DuckDBPyConnection	Open connection to the target database.	required
rebuild	bool	Drop and recreate every table of this schema when the stored version differs. Tables that are not part of the schema, such as a tagsdb schema holding annotations, are left alone.	required

Raises:

Type	Description
TargetError	When the stored version differs and rebuild is false.

Source code in packages/agent-history/src/pdum/agent_history/targets.py

def ensure_schema(connection: duckdb.DuckDBPyConnection, *, rebuild: bool) -> None:
    """Create the tables and views, or replace them when the version differs and a rebuild is allowed.

    Parameters
    ----------
    connection : duckdb.DuckDBPyConnection
        Open connection to the target database.
    rebuild : bool
        Drop and recreate every table of this schema when the stored version differs. Tables that
        are not part of the schema, such as a ``tagsdb`` schema holding annotations, are left alone.

    Raises
    ------
    TargetError
        When the stored version differs and ``rebuild`` is false.
    """
    stored = schema_version(connection)
    if stored is not None and stored != schema.SCHEMA_VERSION:
        if not rebuild:
            raise TargetError(
                f"target holds schema version {stored}, this ingester writes {schema.SCHEMA_VERSION}; "
                "run with --full to rebuild it"
            )
        for view in schema.VIEWS:
            connection.execute(f"DROP VIEW IF EXISTS {view}")
        for table in schema.COLUMNS:
            connection.execute(f"DROP TABLE IF EXISTS {table}")
    schema.create(connection)

schema_version(connection)

Return the schema version a database declares in meta, or None when it has none.

Source code in packages/agent-history/src/pdum/agent_history/targets.py

def schema_version(connection: duckdb.DuckDBPyConnection) -> int | None:
    """Return the schema version a database declares in ``meta``, or None when it has none."""
    tables = {row[0] for row in connection.execute("select table_name from duckdb_tables()").fetchall()}
    if "meta" not in tables:
        return None
    row = connection.execute("select value from meta where key = 'schema_version'").fetchone()
    return int(row[0]) if row and row[0] is not None else None

What a source's files looked like when they were last read, so a run can do only the new work.

The unit of incremental work is one transcript file: a Claude session or sub-agent file, a Codex rollout with its catalogue row. Files are append-only, so size, modification time and a small fingerprint (the catalogue's updated_at for Codex) decide whether a unit changed. A unit whose file has vanished is marked gone and its rows stay: the agents' own stores forget, this one does not. Removing history is an explicit command, never a side effect of a run.

LedgerRow dataclass

One stored ledger entry.

Source code in packages/agent-history/src/pdum/agent_history/ledger.py

@dataclass
class LedgerRow:
    """One stored ledger entry."""

    unit_id: str
    path: str | None
    bytes: int | None
    mtime: datetime | None
    fingerprint: str | None
    status: str
    thread_ids: list[str]
    first_seen_at: datetime | None

Plan dataclass

Which units a run reads, leaves alone or marks gone.

Source code in packages/agent-history/src/pdum/agent_history/ledger.py

@dataclass
class Plan:
    """Which units a run reads, leaves alone or marks gone."""

    changed: dict[str, Unit]
    unchanged: dict[str, Unit]
    gone: dict[str, LedgerRow]

    @property
    def counts(self) -> dict[str, int]:
        return {"changed": len(self.changed), "unchanged": len(self.unchanged), "gone": len(self.gone)}

Unit dataclass

One file as found on disk during a scan.

Parameters:

Name	Type	Description	Default
unit_id	str	Stable id, the thread id the file is about (session or session:agent).	required
path	Path or None	The file; None for a catalogue-only Codex thread without a rollout.	required
bytes	int	File size.	0
mtime	datetime or None	Modification time as naive UTC.	None
fingerprint	str or None	Extra change signal, such as the catalogue's updated_at for Codex threads.	None
extra	dict	Anything the ingester needs later, such as a sub-agent file's parent session.	dict()

Source code in packages/agent-history/src/pdum/agent_history/ledger.py

@dataclass
class Unit:
    """One file as found on disk during a scan.

    Parameters
    ----------
    unit_id : str
        Stable id, the thread id the file is about (``session`` or ``session:agent``).
    path : Path or None
        The file; None for a catalogue-only Codex thread without a rollout.
    bytes : int
        File size.
    mtime : datetime or None
        Modification time as naive UTC.
    fingerprint : str or None
        Extra change signal, such as the catalogue's ``updated_at`` for Codex threads.
    extra : dict
        Anything the ingester needs later, such as a sub-agent file's parent session.
    """

    unit_id: str
    path: Path | None
    bytes: int = 0
    mtime: datetime | None = None
    fingerprint: str | None = None
    extra: dict[str, Any] = field(default_factory=dict)

    @classmethod
    def from_path(cls, unit_id: str, path: Path, fingerprint: str | None = None, **extra: Any) -> Unit:
        stat = path.stat()
        return cls(
            unit_id,
            path,
            bytes=stat.st_size,
            mtime=datetime.fromtimestamp(stat.st_mtime, tz=UTC).replace(tzinfo=None).replace(microsecond=0),
            fingerprint=fingerprint,
            extra=extra,
        )

plan(scan, ledger, *, full)

Compare a scan with the ledger.

Parameters:

Name	Type	Description	Default
scan	dict	Units found on disk, by id.	required
ledger	dict	Stored entries, by id.	required
full	bool	Treat every scanned unit as changed.	required

Source code in packages/agent-history/src/pdum/agent_history/ledger.py

def plan(scan: dict[str, Unit], ledger: dict[str, LedgerRow], *, full: bool) -> Plan:
    """Compare a scan with the ledger.

    Parameters
    ----------
    scan : dict
        Units found on disk, by id.
    ledger : dict
        Stored entries, by id.
    full : bool
        Treat every scanned unit as changed.
    """
    changed: dict[str, Unit] = {}
    unchanged: dict[str, Unit] = {}
    for unit_id, unit in scan.items():
        stored = ledger.get(unit_id)
        same = (
            stored is not None
            and stored.status == PRESENT
            and stored.bytes == unit.bytes
            and stored.mtime == unit.mtime
            and (stored.fingerprint or None) == (unit.fingerprint or None)
            and (stored.path or None) == (str(unit.path) if unit.path else None)
        )
        if full or not same:
            changed[unit_id] = unit
        else:
            unchanged[unit_id] = unit
    gone = {unit_id: row for unit_id, row in ledger.items() if unit_id not in scan and row.status == PRESENT}
    return Plan(changed, unchanged, gone)

read(connection, source_id)

Return the stored ledger for one source, empty when the table is absent.

Source code in packages/agent-history/src/pdum/agent_history/ledger.py

def read(connection: duckdb.DuckDBPyConnection, source_id: str) -> dict[str, LedgerRow]:
    """Return the stored ledger for one source, empty when the table is absent."""
    tables = {row[0] for row in connection.execute("select table_name from duckdb_tables()").fetchall()}
    if "ingest_ledger" not in tables:
        return {}
    rows = connection.execute(
        "select unit_id, path, bytes, mtime, fingerprint, status, thread_ids, first_seen_at "
        "from ingest_ledger where source_id = ?",
        [source_id],
    ).fetchall()
    return {
        row[0]: LedgerRow(row[0], row[1], row[2], row[3], row[4], row[5], json.loads(row[6] or "[]"), row[7])
        for row in rows
    }

rows(source_id, run_id, plan_, ledger, thread_ids, now)

Build the ledger rows a run writes: changed units as present, vanished units as gone.

Parameters:

Name	Type	Description	Default
thread_ids	dict	Thread ids each changed unit produced, so a later replacement can delete exactly those.	required

Source code in packages/agent-history/src/pdum/agent_history/ledger.py

def rows(
    source_id: str,
    run_id: str,
    plan_: Plan,
    ledger: dict[str, LedgerRow],
    thread_ids: dict[str, list[str]],
    now: datetime,
) -> list[tuple]:
    """Build the ledger rows a run writes: changed units as present, vanished units as gone.

    Parameters
    ----------
    thread_ids : dict
        Thread ids each changed unit produced, so a later replacement can delete exactly those.
    """
    out: list[tuple] = []
    for unit_id, unit in plan_.changed.items():
        stored = ledger.get(unit_id)
        out.append(
            schema.row(
                "ingest_ledger",
                source_id=source_id,
                unit_id=unit_id,
                path=str(unit.path) if unit.path else None,
                bytes=unit.bytes,
                mtime=unit.mtime,
                fingerprint=unit.fingerprint,
                status=PRESENT,
                thread_ids=json.dumps(thread_ids.get(unit_id) or [unit_id]),
                first_seen_at=(stored.first_seen_at if stored and stored.first_seen_at else now),
                last_seen_at=now,
                last_ingested_at=now,
                run_id=run_id,
            )
        )
    for unit_id, stored in plan_.gone.items():
        out.append(
            schema.row(
                "ingest_ledger",
                source_id=source_id,
                unit_id=unit_id,
                path=stored.path,
                bytes=stored.bytes,
                mtime=stored.mtime,
                fingerprint=stored.fingerprint,
                status=GONE,
                thread_ids=json.dumps(stored.thread_ids),
                first_seen_at=stored.first_seen_at,
                last_seen_at=stored.first_seen_at,
                last_ingested_at=None,
                run_id=run_id,
            )
        )
    return out

Facts that look across threads, computed from the store after a source's threads are written.

Parsing one file decides what that file owns. Which file continued which, and which root a thread belongs to, need every thread of a source, so the ingesters record per-thread links while parsing (thread_links) and this module turns them into fork_edges and the forked_from_id, lineage_id and lineage_depth columns of threads. It runs on the target after every publish, over metadata only, so an incremental run that replaced three files recomputes a source's lineage in milliseconds.

Link kinds

producer A Claude file copied records messages produced by target_thread_id; ordinal is the order of first appearance, ts the last copied message. The last producer is the parent. continued-in A Claude session's own record naming the session that continued it. fork-context-ref A Claude sub-agent referencing its session's context; detail holds the parent's last uuid. forked-from A Codex rollout's session_meta naming the thread it was forked from.

dedupe_responses(connection, source_id)

Mark responses that several unmarked files of one source claim, keeping the earliest-started thread's.

Parallel sub-agents launched from one call can begin with the same responses, and files written without markers cannot say so themselves. After publishing, any response id that more than one thread of a source holds as its own stays with the thread that started first (ties broken by thread id) and the other copies get origin = 'copied-context', so usage counts them once and the duplicate check stays at zero. One pass settles every group.

Returns:

Type	Description
int	Number of response rows marked as copies.

Source code in packages/agent-history/src/pdum/agent_history/derive.py

def dedupe_responses(connection: duckdb.DuckDBPyConnection, source_id: str) -> int:
    """Mark responses that several unmarked files of one source claim, keeping the earliest-started thread's.

    Parallel sub-agents launched from one call can begin with the same responses, and files
    written without markers cannot say so themselves. After publishing, any response id that more
    than one thread of a source holds as its own stays with the thread that started first (ties
    broken by thread id) and the other copies get ``origin = 'copied-context'``, so usage counts
    them once and the duplicate check stays at zero. One pass settles every group.

    Returns
    -------
    int
        Number of response rows marked as copies.
    """
    before = connection.execute(
        "select count(*) from responses where source_id = ? and origin = 'copied-context'", [source_id]
    ).fetchone()[0]
    connection.execute(
        """
        UPDATE responses r SET origin = 'copied-context'
        FROM (
            SELECT n.thread_id, n.response_id
            FROM responses n
            JOIN threads hn ON hn.thread_id = n.thread_id AND hn.source_id = n.source_id
            JOIN responses o ON o.source_id = n.source_id AND o.response_id = n.response_id
                 AND o.thread_id <> n.thread_id AND o.origin <> 'copied-context'
            JOIN threads ho ON ho.thread_id = o.thread_id AND ho.source_id = o.source_id
            WHERE n.source_id = ? AND n.origin <> 'copied-context'
              AND (ho.created_at < hn.created_at
                   OR (ho.created_at IS NOT DISTINCT FROM hn.created_at AND ho.thread_id < hn.thread_id))
        ) d
        WHERE r.source_id = ? AND r.thread_id = d.thread_id AND r.response_id = d.response_id
        """,
        [source_id, source_id],
    )
    after = connection.execute(
        "select count(*) from responses where source_id = ? and origin = 'copied-context'", [source_id]
    ).fetchone()[0]
    return int(after - before)

derive_lineage(connection, source_id, source)

Rebuild fork_edges and the lineage columns of threads for one source.

Parameters:

Name	Type	Description	Default
connection	DuckDBPyConnection	Connection to the database holding the source's threads and links.	required
source_id	str	Which source to recompute.	required
source	str	The source value written to new edge rows.	required

Returns:

Type	Description
dict	edges written, threads updated and any notes (cycles).

Source code in packages/agent-history/src/pdum/agent_history/derive.py

def derive_lineage(connection: duckdb.DuckDBPyConnection, source_id: str, source: str) -> dict[str, Any]:
    """Rebuild ``fork_edges`` and the lineage columns of ``threads`` for one source.

    Parameters
    ----------
    connection : duckdb.DuckDBPyConnection
        Connection to the database holding the source's threads and links.
    source_id : str
        Which source to recompute.
    source : str
        The ``source`` value written to new edge rows.

    Returns
    -------
    dict
        ``edges`` written, ``threads`` updated and any ``notes`` (cycles).
    """
    threads: dict[str, dict[str, Any]] = {}
    for thread_id, parent, forked, created in connection.execute(
        "select thread_id, parent_thread_id, forked_from_id, created_at from threads where source_id = ?",
        [source_id],
    ).fetchall():
        threads[thread_id] = {
            "thread_id": thread_id,
            "parent_thread_id": parent,
            "forked_from_id": forked if parent else None,
            "created_at": created,
        }
    links: dict[str, list[tuple]] = defaultdict(list)
    for row in connection.execute(
        "select thread_id, kind, target_thread_id, ordinal, records, ts, detail from thread_links "
        "where source_id = ? order by thread_id, kind, ordinal",
        [source_id],
    ).fetchall():
        links[row[0]].append(row)

    edges: list[tuple] = []
    edge_index: dict[str, int] = {}

    def edge(parent: str, child: str, **values: Any) -> None:
        edge_index[child] = len(edges)
        edges.append(
            schema.row(
                "fork_edges",
                parent_thread_id=parent,
                child_thread_id=child,
                source_id=source_id,
                source=source,
                origin="derived",
                **{"ambiguous": False, **values},
            )
        )

    for thread_id, rows in links.items():
        record = threads.get(thread_id)
        producers = [row for row in rows if row[1] == "producer"]
        if producers and record is not None and not record["parent_thread_id"]:
            parent = producers[-1][2]
            record["forked_from_id"] = parent
            edge(
                parent,
                thread_id,
                kind="continuation",
                evidence="marker",
                inherited_records=sum(int(row[4] or 0) for row in producers),
                fork_point_ts=max((row[5] for row in producers if row[5] is not None), default=None),
                child_started_at=record["created_at"],
            )
        for row in rows:
            if row[1] == "fork-context-ref" and record is not None and record["parent_thread_id"]:
                detail = json.loads(row[6] or "{}")
                parent = row[2] or record["parent_thread_id"]
                record["forked_from_id"] = parent
                fork_point = None
                last_uuid = detail.get("parentLastUuid")
                if last_uuid:
                    found = connection.execute(
                        "select created_at from items where source_id = ? and thread_id = ? and uuid = ? "
                        "order by rollout_ordinal limit 1",
                        [source_id, parent, last_uuid],
                    ).fetchone()
                    fork_point = found[0] if found else None
                edge(
                    parent,
                    thread_id,
                    kind="subagent-fork",
                    evidence="fork-context-ref",
                    inherited_records=row[4],
                    fork_point_ts=fork_point,
                    child_started_at=record["created_at"],
                )
            elif row[1] == "forked-from" and record is not None:
                parent = row[2]
                if parent and parent != thread_id:
                    record["forked_from_id"] = parent
                    edge(
                        parent,
                        thread_id,
                        kind="fork",
                        evidence="rollout",
                        child_started_at=record["created_at"],
                    )
    for thread_id, rows in links.items():
        for row in rows:
            if row[1] != "continued-in" or not row[2]:
                continue
            successor = row[2]
            index = edge_index.get(successor)
            if index is not None and edges[index][0] == thread_id:
                evidence_at = [name for name, _ in schema.COLUMNS["fork_edges"]].index("evidence")
                current = edges[index]
                edges[index] = (
                    current[:evidence_at] + (f"{current[evidence_at]}+continued-in",) + current[evidence_at + 1 :]
                )
                continue
            child = threads.get(successor)
            edge(
                thread_id,
                successor,
                kind="continuation",
                evidence="continued-in",
                child_started_at=child["created_at"] if child else row[5],
            )
            if child is not None and not child.get("forked_from_id"):
                child["forked_from_id"] = thread_id
    notes = assign_lineage(threads)

    connection.execute("DELETE FROM fork_edges WHERE source_id = ?", [source_id])
    schema.insert(connection, "fork_edges", edges)
    updates = [
        (record.get("forked_from_id"), record["lineage_id"], record["lineage_depth"], thread_id)
        for thread_id, record in threads.items()
    ]
    if updates:
        connection.register(
            "_lineage_updates",
            _as_table(updates),
        )
        connection.execute(
            "UPDATE threads SET forked_from_id = u.forked_from_id, lineage_id = u.lineage_id, "
            "lineage_depth = u.lineage_depth FROM _lineage_updates u "
            "WHERE threads.thread_id = u.thread_id AND threads.source_id = ?",
            [source_id],
        )
        connection.unregister("_lineage_updates")
    return {"edges": len(edges), "threads": len(updates), "notes": list(notes.values())}