feat(ui): complete postgres phase1 persistence

coding-agent/.opencode/prompts/postgres-phase1-completion-task-packet.md

@@ -0,0 +1,45 @@
+Goal:
+Complete Phase 1 PostgreSQL persistence for the HTTP/UI session stack by adding PG checkpoint storage and config-driven default persistence wiring for session metadata, tapes, and checkpoints.
+
+Scope:
+- Add `PGCheckpointStore` in `agentkit.storage.pg` with matching protocol tests.
+- Add a synchronous UI PostgreSQL session metadata store and one app-layer persistence factory for default SessionManager/HTTP wiring.
+- Update config defaults/tests so HTTP startup and SessionManager defaults can use file or PostgreSQL storage through one shared path.
+
+Out of scope:
+- Phase 2 ownership, fencing, approval routing, or cross-instance event routing.
+- New migration tooling or schema-management framework beyond the repo's existing inline schema creation pattern.
+
+Context:
+- ADRs:
+  - `docs/adr/0012-complete-phase1-postgresql-http-session-persistence.md`
+  - `docs/adr/0001-checkpoint-captures-serialized-tape-and-plugin-state.md`
+  - `docs/adr/0003-http-sessions-use-one-stable-tape-timeline.md`
+  - `docs/adr/0005-checkpoint-restore-uses-truncate-rollback.md`
+- Relevant files:
+  - `src/agentkit/storage/pg.py`
+  - `src/coding_agent/ui/session_store.py`
+  - `src/coding_agent/ui/session_manager.py`
+  - `src/coding_agent/ui/http_server.py`
+  - `src/coding_agent/agent.toml`
+  - `tests/agentkit/storage/test_pg.py`
+  - `tests/ui/test_session_persistence.py`
+  - `tests/ui/test_session_manager_public_api.py`
+  - `tests/ui/test_http_server.py`
+
+Target tests:
+- `uv run pytest tests/agentkit/storage/test_pg.py -k "checkpoint or session or tape" -v`
+- `uv run pytest tests/ui/test_session_persistence.py tests/ui/test_session_manager_public_api.py tests/ui/test_http_server.py -k "pg or checkpoint or persistence" -v`
+- `uv run pytest tests/ -v`
+
+Loop policy:
+- Engineer implements the smallest correct change and runs the target tests.
+- Reviewer reviews only the resulting diff and affected tests.
+- Reviewer reports only P1/P2 findings.
+- Engineer fixes only accepted P1/P2 findings and reruns the same target tests.
+- Verifier reruns the exact target tests and reports pass/fail only.
+
+Stop conditions:
+- At most one review/fix/retest cycle.
+- Escalate architectural redirection or scope expansion to the human.
+- Ignore non-blocking optimization suggestions.

coding-agent/docs/adr/0012-complete-phase1-postgresql-http-session-persistence.md

@@ -0,0 +1,53 @@
+# ADR-0012: Complete Phase 1 PostgreSQL HTTP session persistence
+
+**Status**: Proposed
+**Date**: 2026-04-17
+
+## Context
+
+The repository already includes reusable PostgreSQL persistence primitives for session metadata and tapes in `agentkit.storage.pg`, and the HTTP/UI session stack already has stable tape and checkpoint semantics backed by tests. But the Phase 1 goal for HTTP session persistence is still incomplete on `main`.
+
+`SessionManager` still defaults to the local JSONL tape store and filesystem checkpoint store, `http_server.py` still constructs `SessionManager()` directly with those defaults, and the UI session metadata layer still only provides in-memory and Redis-backed implementations. There is also no PostgreSQL `CheckpointStore` implementation in `agentkit`, which means the current HTTP checkpoint path cannot persist checkpoint snapshots/catalogs in PostgreSQL even when tape persistence uses PostgreSQL.
+
+Phase 1 is still explicitly single-instance and keeps active runtime state (`task`, `approval_event`, `event_queues`, `runtime_pipeline`, `runtime_ctx`, `runtime_adapter`) local-authoritative. This ADR only closes the durable storage gap for restart-safe metadata, stable tape history, and checkpoint snapshots/catalogs.
+
+## Decision
+
+Complete Phase 1 by adding PostgreSQL-backed checkpoint persistence in `agentkit` and one config-driven app-layer persistence path for the HTTP/UI session stack.
+
+Specifically:
+
+- Add `PGCheckpointStore` to `agentkit.storage.pg` implementing the existing `CheckpointStore` protocol.
+- Add a synchronous PostgreSQL-backed UI session metadata store in `coding_agent.ui.session_store` rather than widening `SessionManager` to the async `agentkit` session protocol.
+- Add a small app-layer persistence factory in `coding_agent.ui` that builds the default session metadata store, tape store, checkpoint store, and shutdown hook from the existing flat `[storage]` config in `src/coding_agent/agent.toml`.
+- Make both `SessionManager()` defaults and HTTP server startup use that same persistence factory instead of hard-coded JSONL/filesystem/in-memory defaults.
+- Keep runtime execution ownership, approval routing, and event routing out of scope for this ADR and Phase 1.
+
+## Alternatives Rejected
+
+- Reuse the async `agentkit.PGSessionStore` directly inside `SessionManager` — rejected because it would widen a synchronous UI/session-manager boundary into a larger refactor.
+- Keep checkpoints on the filesystem while only moving tapes to PostgreSQL — rejected because checkpoint restore belongs to the same stable tape timeline and Phase 1 needs durable checkpoint snapshots/catalogs.
+- Add PostgreSQL support by sprinkling separate wiring logic across `SessionManager`, `http_server.py`, and REPL startup — rejected because Phase 1 needs one config-driven default path, not divergent per-entrypoint behavior.
+- Expand into Phase 2 owner/lease/fencing work while touching persistence — rejected because Phase 1 remains single-instance and local-authoritative for active runtime state.
+
+## Acceptance Criteria
+
+- [ ] PostgreSQL checkpoint persistence is covered by tests in `tests/agentkit/storage/test_pg.py`, including round-trip snapshot persistence and ordered listing by tape.
+- [ ] Config-driven persistence bundle construction from `[storage]` configuration is covered by tests in `tests/coding_agent/plugins/test_storage_factory.py`.
+- [ ] `SessionManager` default persistence wiring is covered by tests in `tests/ui/test_session_manager_public_api.py` and `tests/ui/test_session_persistence.py`.
+- [ ] HTTP server startup uses the configured persistence bundle for the global session manager, covered by tests in `tests/ui/test_http_server.py`.
+- [ ] `uv run pytest tests/agentkit/storage/test_pg.py tests/coding_agent/plugins/test_storage_factory.py tests/ui/test_session_persistence.py tests/ui/test_session_manager_public_api.py tests/ui/test_http_server.py -v`
+
+## References
+
+- `src/agentkit/storage/pg.py`
+- `src/agentkit/storage/protocols.py`
+- `src/agentkit/checkpoint/service.py`
+- `src/coding_agent/ui/session_store.py`
+- `src/coding_agent/ui/session_manager.py`
+- `src/coding_agent/ui/http_server.py`
+- `src/coding_agent/agent.toml`
+- `tests/agentkit/storage/test_pg.py`
+- `tests/ui/test_session_persistence.py`
+- `tests/ui/test_session_manager_public_api.py`
+- `tests/ui/test_http_server.py`

coding-agent/src/agentkit/storage/pg.py

@@ -4,8 +4,11 @@
 import importlib
 import json
 from collections.abc import Awaitable, Callable
+from datetime import datetime
 from typing import Final, Protocol, cast
 
+from agentkit.checkpoint.models import CheckpointMeta, CheckpointSnapshot
+
 
 class AsyncPGPool(Protocol):
     async def execute(self, query: str, *args: object) -> str: ...
@@ -31,6 +34,17 @@ async def acquire(self) -> LockConnection: ...
     async def release(self, connection: LockConnection) -> None: ...
 
 
+class CodecConnection(Protocol):
+    async def set_type_codec(
+        self,
+        typename: str,
+        *,
+        encoder: Callable[[object], str],
+        decoder: Callable[[str], object],
+        schema: str,
+    ) -> None: ...
+
+
 PoolFactory = Callable[..., Awaitable[AsyncPGPool]]
 
 
@@ -59,6 +73,7 @@ async def get_pool(self) -> AsyncPGPool:
                         dsn=self._dsn,
                         min_size=self._min_size,
                         max_size=self._max_size,
+                        init=_init_connection_codecs,
                     )
         return self._pool
 
@@ -110,7 +125,7 @@ async def _ensure_schema(self) -> AsyncPGPool:
 
     async def save_session(self, session_id: str, data: dict[str, object]) -> None:
         pool = await self._ensure_schema()
-        _ = await pool.execute(self._UPSERT_SQL, session_id, json.dumps(data))
+        _ = await pool.execute(self._UPSERT_SQL, session_id, data)
 
     async def load_session(self, session_id: str) -> dict[str, object] | None:
         pool = await self._ensure_schema()
@@ -150,13 +165,21 @@ class PGTapeStore:
         PRIMARY KEY (tape_id, seq)
     )
     """
-    _MAX_SEQ_SQL: Final[str] = (
-        "SELECT MAX(seq) AS max_seq FROM agent_tapes WHERE tape_id = $1"
-    )
     _INSERT_SQL: Final[str] = """
+    WITH advisory_lock AS (
+        SELECT pg_advisory_xact_lock(hashtext($1))
+    ),
+    max_seq AS (
+        SELECT COALESCE(MAX(seq), -1) AS value
+        FROM agent_tapes
+        WHERE tape_id = $1
+    )
     INSERT INTO agent_tapes (tape_id, seq, entry)
-    SELECT $1, seq, entry::jsonb
-    FROM unnest($2::int[], $3::text[]) AS batch(seq, entry)
+    SELECT
+        $1,
+        (SELECT value FROM max_seq) + batch.ordinality,
+        batch.entry::jsonb
+    FROM advisory_lock, unnest($2::text[]) WITH ORDINALITY AS batch(entry, ordinality)
     """
     _LOAD_SQL: Final[str] = (
         "SELECT entry FROM agent_tapes WHERE tape_id = $1 ORDER BY seq"
@@ -182,25 +205,8 @@ async def save(self, tape_id: str, entries: list[dict[str, object]]) -> None:
             return
 
         pool = await self._ensure_schema()
-        row = await pool.fetchrow(self._MAX_SEQ_SQL, tape_id)
-        max_seq = -1
-        if row is not None:
-            raw_max_seq = row.get("max_seq")
-            if raw_max_seq is None:
-                max_seq = -1
-            elif isinstance(raw_max_seq, int):
-                max_seq = raw_max_seq
-            else:
-                raise TypeError("postgres tape row must include int max_seq or None")
-
-        seq_values = [max_seq + offset for offset, _ in enumerate(entries, start=1)]
         payload_values = [json.dumps(entry) for entry in entries]
-        _ = await pool.execute(
-            self._INSERT_SQL,
-            tape_id,
-            seq_values,
-            payload_values,
-        )
+        _ = await pool.execute(self._INSERT_SQL, tape_id, payload_values)
 
     async def load(self, tape_id: str) -> list[dict[str, object]]:
         pool = await self._ensure_schema()
@@ -231,6 +237,94 @@ async def truncate(self, tape_id: str, keep: int) -> None:
         _ = await pool.execute(self._TRUNCATE_SQL, tape_id, keep)
 
 
+class PGCheckpointStore:
+    _CREATE_TABLE_SQL: Final[str] = """
+    CREATE TABLE IF NOT EXISTS agent_checkpoints (
+        checkpoint_id TEXT PRIMARY KEY,
+        tape_id TEXT NOT NULL,
+        meta JSONB NOT NULL,
+        entries JSONB NOT NULL,
+        plugin_states JSONB NOT NULL,
+        extra JSONB NOT NULL
+    )
+    """
+    _INSERT_SQL: Final[str] = """
+    INSERT INTO agent_checkpoints (
+        checkpoint_id,
+        tape_id,
+        meta,
+        entries,
+        plugin_states,
+        extra
+    )
+    VALUES ($1, $2, $3::jsonb, $4::jsonb, $5::jsonb, $6::jsonb)
+    ON CONFLICT (checkpoint_id) DO UPDATE SET
+        tape_id = EXCLUDED.tape_id,
+        meta = EXCLUDED.meta,
+        entries = EXCLUDED.entries,
+        plugin_states = EXCLUDED.plugin_states,
+        extra = EXCLUDED.extra
+    """
+    _LOAD_SQL: Final[str] = (
+        "SELECT meta, entries, plugin_states, extra FROM agent_checkpoints "
+        "WHERE checkpoint_id = $1"
+    )
+    _LIST_SQL: Final[str] = (
+        "SELECT meta FROM agent_checkpoints WHERE tape_id = $1 "
+        "ORDER BY (meta->>'created_at')::timestamptz"
+    )
+    _DELETE_SQL: Final[str] = "DELETE FROM agent_checkpoints WHERE checkpoint_id = $1"
+
+    def __init__(self, *, pool: PGPool) -> None:
+        self._pool: PGPool = pool
+        self._schema_ready: bool = False
+
+    async def _ensure_schema(self) -> AsyncPGPool:
+        pool = await self._pool.get_pool()
+        if not self._schema_ready:
+            _ = await pool.execute(self._CREATE_TABLE_SQL)
+            self._schema_ready = True
+        return pool
+
+    async def save(self, snapshot: CheckpointSnapshot) -> None:
+        pool = await self._ensure_schema()
+        meta = snapshot.meta
+        meta_payload = {
+            "checkpoint_id": meta.checkpoint_id,
+            "tape_id": meta.tape_id,
+            "session_id": meta.session_id,
+            "entry_count": meta.entry_count,
+            "window_start": meta.window_start,
+            "created_at": meta.created_at.isoformat(),
+            "label": meta.label,
+        }
+        _ = await pool.execute(
+            self._INSERT_SQL,
+            meta.checkpoint_id,
+            meta.tape_id,
+            meta_payload,
+            list(snapshot.tape_entries),
+            snapshot.plugin_states,
+            snapshot.extra,
+        )
+
+    async def load(self, checkpoint_id: str) -> CheckpointSnapshot | None:
+        pool = await self._ensure_schema()
+        row = await pool.fetchrow(self._LOAD_SQL, checkpoint_id)
+        if row is None:
+            return None
+        return _snapshot_from_row(row)
+
+    async def list_by_tape(self, tape_id: str) -> list[CheckpointMeta]:
+        pool = await self._ensure_schema()
+        rows = await pool.fetch(self._LIST_SQL, tape_id)
+        return [_meta_from_raw(_required_dict(row, "meta")) for row in rows]
+
+    async def delete(self, checkpoint_id: str) -> None:
+        pool = await self._ensure_schema()
+        _ = await pool.execute(self._DELETE_SQL, checkpoint_id)
+
+
 class PGSessionLock:
     def __init__(self, *, pool: LockPool) -> None:
         self._pool: LockPool = pool
@@ -254,7 +348,10 @@ async def release(self) -> None:
         try:
             _ = await self._conn.execute("SELECT pg_advisory_unlock_all()")
         finally:
-            await self._pool.release(self._conn)
+            pool_release: Awaitable[None] = cast(
+                Awaitable[None], self._pool.release(self._conn)
+            )
+            await pool_release
             self._conn = None
 
 
@@ -270,3 +367,81 @@ def _load_asyncpg_pool_factory() -> PoolFactory:
     if not callable(create_pool):
         raise ImportError("asyncpg does not expose create_pool")
     return cast(PoolFactory, create_pool)
+
+
+async def _init_connection_codecs(connection: CodecConnection) -> None:
+    for typename in ("json", "jsonb"):
+        await connection.set_type_codec(
+            typename,
+            encoder=json.dumps,
+            decoder=json.loads,
+            schema="pg_catalog",
+        )
+
+
+def _required_dict(row: dict[str, object], key: str) -> dict[str, object]:
+    value = row.get(key)
+    if not isinstance(value, dict):
+        raise TypeError(f"postgres checkpoint row must include dict {key}")
+    return cast(dict[str, object], value)
+
+
+def _required_list(row: dict[str, object], key: str) -> list[object]:
+    value = row.get(key)
+    if not isinstance(value, list):
+        raise TypeError(f"postgres checkpoint row must include list {key}")
+    return value
+
+
+def _meta_from_raw(raw: dict[str, object]) -> CheckpointMeta:
+    checkpoint_id = raw.get("checkpoint_id")
+    tape_id = raw.get("tape_id")
+    session_id = raw.get("session_id")
+    entry_count = raw.get("entry_count")
+    window_start = raw.get("window_start")
+    created_at = raw.get("created_at")
+    label = raw.get("label")
+
+    if not isinstance(checkpoint_id, str):
+        raise TypeError("postgres checkpoint meta must include string checkpoint_id")
+    if not isinstance(tape_id, str):
+        raise TypeError("postgres checkpoint meta must include string tape_id")
+    if session_id is not None and not isinstance(session_id, str):
+        raise TypeError(
+            "postgres checkpoint meta must include string session_id or None"
+        )
+    if not isinstance(entry_count, int):
+        raise TypeError("postgres checkpoint meta must include int entry_count")
+    if not isinstance(window_start, int):
+        raise TypeError("postgres checkpoint meta must include int window_start")
+    if not isinstance(created_at, str):
+        raise TypeError("postgres checkpoint meta must include string created_at")
+    if label is not None and not isinstance(label, str):
+        raise TypeError("postgres checkpoint meta must include string label or None")
+
+    return CheckpointMeta(
+        checkpoint_id=checkpoint_id,
+        tape_id=tape_id,
+        session_id=session_id,
+        entry_count=entry_count,
+        window_start=window_start,
+        created_at=datetime.fromisoformat(created_at),
+        label=label,
+    )
+
+
+def _snapshot_from_row(row: dict[str, object]) -> CheckpointSnapshot:
+    entries_raw = _required_list(row, "entries")
+    plugin_states = _required_dict(row, "plugin_states")
+    extra = _required_dict(row, "extra")
+    tape_entries: list[dict[str, object]] = []
+    for entry in entries_raw:
+        if not isinstance(entry, dict):
+            raise TypeError("postgres checkpoint entries must contain dict items")
+        tape_entries.append(cast(dict[str, object], entry))
+    return CheckpointSnapshot(
+        meta=_meta_from_raw(_required_dict(row, "meta")),
+        tape_entries=tuple(tape_entries),
+        plugin_states=plugin_states,
+        extra=extra,
+    )

coding-agent/src/coding_agent/agent.toml

@@ -23,6 +23,8 @@ enabled = [
 
 [storage]
 tape_backend = "jsonl"
+http_session_backend = "memory"
+checkpoint_backend = "fs"
 doc_backend = "lancedb"
 
 [storage.paths]