From 58b3cc1a5abc40b696051c19a3fb36a0e9dc28e3 Mon Sep 17 00:00:00 2001 From: CC#3 Kora Runtime Date: Wed, 20 May 2026 04:21:32 -0700 Subject: [PATCH] =?UTF-8?q?feat(KR-2=20ST4):=20chain=20events=20+=20Consti?= =?UTF-8?q?tution=20+=20session=20context=20=E2=80=94=20closes=20KR-2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ST4 wraps KR-2. Recent chain events read, active Constitution revision read, KoraSessionContext shape + assembler, system prompt block §6, and removal of every remaining NotImplementedError stub. Chain event emit is deferred behind ChainEventEmitNotAvailableError until the Sea MCP tool ships (same pattern as the ST3 scratchpad write defer). Read paths (new): * events.py:read_recent_kora_events — direct asyncpg against hivex_foundation.event_log, the substrate's one genuine tenant_id-UUID-keyed table. JOINs hivex_foundation.tenant on clerk_org_id to resolve the workspace_id (Clerk TEXT) → tenant_id. Filters event_type LIKE 'kora.%'. jsonb_pretty(payload) truncated to 300 chars + ellipsis to match the TS-side KoraRecentChainEvent shape. Default LIMIT 50 per spec. * constitution.py:read_active_constitution_revision — asyncpg against kronicle.workspace_constitution_revisions. RLS-GUC-in-transaction (same pattern as policy_registry / scratchpad). No superseded_at column on the table — active = ORDER BY revision_number DESC LIMIT 1. rules_hash BYTEA hex-encoded to match the TS-side constitutionVersionHashFromRulesHash helper so K-6 (Constitution pre-screen middleware) sees the same canonical hex form. Returns None for fresh-workspace bootstrap state. Session context (new): * session_context.py:assemble_session_context — fans out the six load-bearing reads via asyncio.gather. Mirrors TS-side KoraSessionContext shape at packages/sea-mcp-server/src/kora/context-assembler/types.ts:130. * KoraSessionContext dataclass: workspace_id, assembled_at, role_charter, capability_matrix_row, own_scratchpad, cross_agent_scratchpad, recent_chain_events, active_constitution_revision_id, active_constitution_rules_hash. Tuple-typed read collections — immutable end-to-end like TS side. Deferred chain event emit: * events.py:emit_kora_event raises ChainEventEmitNotAvailableError with [kora.isokron.todo] tag + D-kr2-st4-no-chain-emit-mcp-tool. Signature matches the future MCP-backed impl so caller code stays stable when the substrate tool lands. Direct INSERT into event_log forbidden — would skip _emit_chain_event SECDEF and break prev_event_hash / this_event_hash witness chain. Provider finalize: * on_turn_start now gathers 7 reads in one shot (the original 3 from ST2 + the 2 scratchpad reads from ST3 + the 2 ST4 reads). Single asyncio.gather per prefetch for one round-trip latency. * IsoKronMemoryProvider.session_context() returns the typed KoraSessionContext from warm caches (or None when caches are cold). * system_prompt_block extended with §6 "Recent kora.* activity" surfacing the last 5 events as readable bullets; cap is bounded to keep the prompt size sane. * All remaining ABC stubs replaced with real implementations: - on_session_end emits kora.session.ended (deferred + caught) - on_session_switch updates session_id; reset=True flushes all 7 caches so the next turn re-reads against fresh substrate - on_pre_compress returns "" (substrate is the system of record; nothing isokron-side to inject into compression summaries) - on_delegation mirrors as scratchpad reasoning_trail entry + emits kora.handoff.to_claude_pm (both deferred + caught) - save_config no-op (env-var-based config; no native YAML file) - handle_tool_call inherits ABC default (provider has no tools until KR-3 ships iso_node_* / iso_link_*) * _attempt_chain_event_emit helper mirrors _attempt_scratchpad_write's pattern: submit-and-wait, catch the defer-error, log a one-line WARNING tagged with the deviation ID. Successful emits invalidate the events cache. Capability mirror parity fix: * The C2 mirror was 1 cap stale — Sea MCP capability-matrix.ts has added cap_unbless_convention (operator-only, kora: false) since KR-2 ST2 shipped. The parity test fired in this PR's run and caught the drift in both directions. Mirror + counts updated to match: SEA 24 + KORA_BROADER 25 = 49 total; Kora granted=22, denied=27. Parity test now passes against current substrate main (28ff4f78). This is the exact win the C2 parity guard was built for; closing D-kr2-st2 still gated on K-7 substrate dispatch. Tests (23 new): * test_events.py (9): typed shape, default limit 50, custom limit binding, SQL JOINs tenant on clerk_org_id + LIKE filter, payload truncation at 300 chars + ellipsis, short payload pass-through, empty payload, datetime occurred_at ISO encoding, deferred emit carries the tag. * test_constitution.py (8): typed shape with hex-encoded rules_hash, None for fresh workspace, RLS-GUC ordering, SQL uses revision_number DESC (NOT superseded_at), hex encoder accepts bytes / bytearray / memoryview / str / rejects unknown types. * test_session_context.py (3): fans out all 6 reads, None pair on no Constitution row, default limits. * test_provider_end_to_end.py (2): full lifecycle walk (initialize → on_turn_start → system_prompt_block → sync_turn → on_memory_write → on_delegation → on_pre_compress → on_session_switch → on_session_end → reset → shutdown). Asserts no NotImplementedError surfaces anywhere; deferred-emit / deferred- scratchpad warnings logged but caught. Cold-cache session_context returns None. ST1 skeleton test updated: parametrize replaced with test_no_stub_methods_remain_after_st4 verifying each ABC method returns normally; test_handle_tool_call_unsupported_tool_raises_ with_provider_name verifies the ABC's clear-error path is preserved. Local gates: * ty check — 7,337 diagnostics, same as ST3 baseline (zero-delta). * pytest tests/plugins/memory/ — 248/248 passing (23 new ST4 + 64 pre-ST4 isokron + 161 other memory providers). * Full suite via xdist (-n auto): 24,570 passed / 143 failed / 129 skipped. Δ vs ST3 merge baseline (24,551/143/129): +19 passed, ±0 failed. Failures still concentrated in tests/tools/* + tests/tui_gateway/* xdist isolation noise; none touch plugins/memory/isokron/. Rule-6: * BUILD_DEVIATIONS.md gains D-kr2-st4-no-chain-emit-mcp-tool under Open with grep snapshot of available kora__* tools (still 3) and exact closure condition. * README "Operator pitfalls" gains 3 entries: chain emit deferred + tag to grep, event_log tenant_id-keying exception, Constitution revisions table has no superseded_at column. * All [kora.isokron.todo] tags retained on deferred surfaces. KR-2 milestone closed. Standing by for KR-3 dispatch + K-7 + K-8 + K-9 (or equivalent) substrate dispatches to swap the three C2 / deferred surfaces. Co-Authored-By: Claude Opus 4.7 (1M context) --- BUILD_DEVIATIONS.md | 35 ++ plugins/memory/isokron/README.md | 10 +- .../isokron/capability_matrix_mirror.py | 1 + plugins/memory/isokron/constitution.py | 111 ++++++ plugins/memory/isokron/events.py | 192 +++++++++ plugins/memory/isokron/provider.py | 375 +++++++++++++++--- plugins/memory/isokron/session_context.py | 111 ++++++ .../memory/test_capability_matrix_parity.py | 6 +- tests/plugins/memory/test_constitution.py | 157 ++++++++ tests/plugins/memory/test_events.py | 186 +++++++++ .../memory/test_isokron_provider_skeleton.py | 60 +-- .../memory/test_provider_end_to_end.py | 314 +++++++++++++++ tests/plugins/memory/test_reads.py | 24 +- tests/plugins/memory/test_session_context.py | 176 ++++++++ 14 files changed, 1670 insertions(+), 88 deletions(-) create mode 100644 plugins/memory/isokron/constitution.py create mode 100644 plugins/memory/isokron/events.py create mode 100644 plugins/memory/isokron/session_context.py create mode 100644 tests/plugins/memory/test_constitution.py create mode 100644 tests/plugins/memory/test_events.py create mode 100644 tests/plugins/memory/test_provider_end_to_end.py create mode 100644 tests/plugins/memory/test_session_context.py diff --git a/BUILD_DEVIATIONS.md b/BUILD_DEVIATIONS.md index 11306ed65bf6..ff9c9ae3041a 100644 --- a/BUILD_DEVIATIONS.md +++ b/BUILD_DEVIATIONS.md @@ -17,6 +17,41 @@ Format: ## Open +### D-kr2-st4-no-chain-emit-mcp-tool + +- **Bucket**: KR-2 ST4 (chain event emission + recent events read + finalize) +- **Why**: Spec § ST4 § 1 mandates chain events go through a Sea MCP + tool (working name `kora__append_event`) — direct INSERT into + `hivex_foundation.event_log` is forbidden because it would skip the + substrate's `_emit_chain_event` SECDEF (which sets `prev_event_hash` / + `this_event_hash` to maintain chain witness integrity). The Sea MCP + server on substrate main `28ff4f78` exposes only + `kora__propose_convention`, `kora__read_escalation_queue`, + `kora__propose_policy_change` — no append-event tool. Same pattern + as the ST3 scratchpad-write deferral. +- **Closes when**: A Sea MCP append-event tool ships (working name + `kora__append_event`; PM coordinates with substrate-team / files + the substrate dispatch — likely K-9 on CC#1's lane, queued behind + K-7 + K-8). When it lands, `events.emit_kora_event` body switches + from `raise ChainEventEmitNotAvailableError()` to + `mcp_client.invoke('kora__append_event', ...)`. Caller signature + stays unchanged — `provider._attempt_chain_event_emit` and every + lifecycle hook that uses it (`sync_turn`, `on_memory_write`, + `on_session_end`, `on_delegation`) keep working without refactor. +- **Guarded by**: + - `plugins/memory/isokron/events.py` — top-of-module `[kora.isokron.todo]` + tag; `ChainEventEmitNotAvailableError` carries the deviation ID in + every raised message. + - `IsoKronMemoryProvider._attempt_chain_event_emit` — catches + `ChainEventEmitNotAvailableError` + logs a one-line WARNING + tagged with the deviation ID and the event_type that was skipped. + Operators grep `D-kr2-st4-no-chain-emit-mcp-tool` in logs. + - `plugins/memory/isokron/README.md` § "Operator pitfalls" — + chain event deferral notice. + - `tests/plugins/memory/test_events.py` — + `test_emit_kora_event_raises_deferred_write_error` asserts the + error message + tag stay correct. + ### D-kr2-st3-no-scratchpad-write-mcp-tool - **Bucket**: KR-2 ST3 (Scratchpad reads + writes) diff --git a/plugins/memory/isokron/README.md b/plugins/memory/isokron/README.md index 9e5f2ace465e..c60c707e7b9f 100644 --- a/plugins/memory/isokron/README.md +++ b/plugins/memory/isokron/README.md @@ -83,11 +83,17 @@ This `README.md` ships with **KR-2 ST1**, which delivers the structural skeleton |---|---| | ST1 | skeleton; config schema; connection plumbing (IO loop, no real handshakes); plugin discovery wiring; smoke tests | | ST2 | reads: `read_active_role_charter` (SHA-256 integrity, asyncpg), `read_kora_capability_row` (C2 Python mirror — see "Operator pitfalls"), `read_kora_policy_registry` (RLS GUC + 31-row sanity, asyncpg); 60s TTL cache wired per-workspace; `system_prompt_block` assembles identity / CAN / CANNOT / active policies / granted caps / Rule-6 honest-label; `on_turn_start` pre-fetches all three in parallel via `asyncio.gather` | -| ST3 (this PR) | scratchpad reads (`read_own_scratchpad`, `read_cross_agent_scratchpad`) against `kronicle.agent_scratchpad_entries` JOINing `public.actor_registry` for `actor_kind` / `display_name`; RLS-GUC-in-transaction; BLAKE3 integrity warn-on-mismatch; cached 60s per workspace. **Writes deferred** behind `ScratchpadWriteNotAvailableError` — see "Operator pitfalls" and `BUILD_DEVIATIONS.md` D-kr2-st3-no-scratchpad-write-mcp-tool. `sync_turn` + `on_memory_write` attempt writes through the deferred surface + catch the error gracefully. | -| ST4 | `kora.*` chain event emission via Sea MCP `append_event`; recent events read from `hivex_foundation.event_log`; E2E test removes the last `NotImplementedError` stubs | +| ST3 | scratchpad reads (`read_own_scratchpad`, `read_cross_agent_scratchpad`) against `kronicle.agent_scratchpad_entries` JOINing `public.actor_registry` for `actor_kind` / `display_name`; RLS-GUC-in-transaction; BLAKE3 integrity warn-on-mismatch; cached 60s per workspace. **Writes deferred** behind `ScratchpadWriteNotAvailableError` — see "Operator pitfalls" and `BUILD_DEVIATIONS.md` D-kr2-st3-no-scratchpad-write-mcp-tool. `sync_turn` + `on_memory_write` attempt writes through the deferred surface + catch the error gracefully. | +| ST4 (this PR) | Recent `kora.*` chain events read against `hivex_foundation.event_log` (the substrate's one tenant_id-UUID-keyed table; JOIN tenant on clerk_org_id). Active Constitution revision read against `kronicle.workspace_constitution_revisions` (no `superseded_at`; ORDER BY revision_number DESC). `KoraSessionContext` shape + assembler mirroring the TS-side types.ts:130 contract. System prompt block extended with §6 "Recent kora.* activity". All six remaining ABC stubs replaced with real implementations. **Chain event emit deferred** behind `ChainEventEmitNotAvailableError` — see "Operator pitfalls" and `BUILD_DEVIATIONS.md` D-kr2-st4-no-chain-emit-mcp-tool. **KR-2 milestone closes.** | ## Operator pitfalls +* **Chain event emission is currently deferred — `kora.*` events are NOT being written to `event_log`.** KR-2 ST4 ships the emit API (`events.emit_kora_event`) but the substrate-side Sea MCP tool (`kora__append_event` or equivalent) doesn't exist yet (substrate main `28ff4f78`). Until it lands, every emit raises `ChainEventEmitNotAvailableError`; `sync_turn` / `on_memory_write` / `on_delegation` / `on_session_end` catch it + log a one-line WARNING tagged `D-kr2-st4-no-chain-emit-mcp-tool`. This is a chain-audit gap — operators inspecting Kora's recent activity via the `system_prompt_block` §6 section will see only events that landed in `event_log` through other paths (e.g. SECDEF-emitted events from `compact_scratchpad`). Direct INSERT into `event_log` is forbidden — it would skip the `_emit_chain_event` SECDEF and break the `prev_event_hash` / `this_event_hash` witness chain. Tracked in `BUILD_DEVIATIONS.md`. + +* **`event_log` is the one genuine `tenant_id UUID`-keyed substrate table.** Every other Kora table (`kora_role_charter`, `kora_policy_registry`, `kronicle.agent_scratchpad_entries`, `kronicle.workspace_constitution_revisions`) is `workspace_id TEXT`-keyed. `read_recent_kora_events` resolves the workspace_id (Clerk `org_*`) to tenant_id via `JOIN hivex_foundation.tenant ON t.clerk_org_id = $1`. If you bypass `events.read_recent_kora_events` and write your own SQL, replicate the JOIN — a `WHERE workspace_id = $1` against `event_log` will fail (no such column on that table). + +* **Constitution revision table has no `superseded_at` column.** Earlier bucket prompts referenced `WHERE superseded_at IS NULL` which would error with "column does not exist". The actual "active" semantic for `kronicle.workspace_constitution_revisions` is `ORDER BY revision_number DESC LIMIT 1` (riding the `idx_constitution_revisions_workspace_current` index). The `read_active_constitution_revision` reader gets this right; if you query the table directly, copy the SQL from `constitution.py:SELECT_ACTIVE_CONSTITUTION_REVISION_SQL`. + * **Scratchpad writes are currently deferred — sessions still run but lose their reasoning trail.** KR-2 ST3 ships the write API (`scratchpad.write_scratchpad_entry`) but the substrate-side Sea MCP tool (`kora__write_agent_scratchpad`) doesn't exist yet (substrate main `a3e77f67`). Until it lands, every write raises `ScratchpadWriteNotAvailableError`; `sync_turn` and `on_memory_write` catch it + log a one-line WARNING tagged with the BUILD_DEVIATIONS ID. Grep `D-kr2-st3-no-scratchpad-write-mcp-tool` in logs to see how often writes are being deferred. Reads (own + cross-agent) work fully. Tracked in `BUILD_DEVIATIONS.md`. The spec is explicit: do NOT bypass with direct INSERT — that would skip `cap_write_agent_scratchpad` authorization + the `approved_event_id` chain event + visibility_scope validation. * **Scratchpad BLAKE3 integrity is warn-only, NOT fail-closed.** Unlike the Role Charter (which raises on hash mismatch), `read_own_scratchpad` and `read_cross_agent_scratchpad` log a WARNING and return the entry on mismatch. Spec § ST3: scratchpad is mutable working memory; refusing to surface a drifted entry would block sessions on transient state. Operators monitoring chain-of-custody should grep `content_hash drift` in logs. diff --git a/plugins/memory/isokron/capability_matrix_mirror.py b/plugins/memory/isokron/capability_matrix_mirror.py index 72db544020f5..b1ee243666c6 100644 --- a/plugins/memory/isokron/capability_matrix_mirror.py +++ b/plugins/memory/isokron/capability_matrix_mirror.py @@ -115,6 +115,7 @@ # Operator-direct admin caps (operator-ONLY) "cap_operator_approve_policy_change": False, "cap_operator_bless_convention": False, + "cap_unbless_convention": False, # operator-only un-bless path "cap_operator_ack_escalation": False, "cap_operator_update_policy": False, } diff --git a/plugins/memory/isokron/constitution.py b/plugins/memory/isokron/constitution.py new file mode 100644 index 000000000000..4c21b746353a --- /dev/null +++ b/plugins/memory/isokron/constitution.py @@ -0,0 +1,111 @@ +"""Active Constitution revision read (KR-2 ST4). + +Reads the workspace's currently-active Constitution revision from +``kronicle.workspace_constitution_revisions`` (foundation/0083). + +**Schema gotcha** verified against the migration on substrate main: + +- The table has NO ``superseded_at`` column. Earlier bucket-prompt + drafts (and PM dispatches) referenced ``WHERE superseded_at IS NULL`` + — that would fail with "column does not exist". The actual "active" + semantic is ``ORDER BY revision_number DESC LIMIT 1``, riding the + ``idx_constitution_revisions_workspace_current`` index. +- ``rules_hash`` is ``BYTEA`` (raw bytes), not pre-hex-encoded. We + hex-encode here to match the TS-side + ``constitutionVersionHashFromRulesHash`` helper so K-3 (TS + context-assembler) and KR-2 (Python provider) emit the same + canonical hex form for K-6's Constitution pre-screen middleware. +- RLS is enabled keyed off ``current_setting('app.current_workspace_id')`` + — same pattern as ``kora_policy_registry`` and ``agent_scratchpad_entries``. + The reader uses the GUC-in-transaction pattern from ST2/ST3. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass +from typing import Any, Optional + +logger = logging.getLogger(__name__) + + +@dataclass(frozen=True, slots=True) +class ActiveConstitutionRevision: + """Active Constitution revision for a workspace. + + Mirrors TS-side ``ActiveConstitutionRevision`` from + ``packages/sea-mcp-server/src/kora/context-assembler/index.ts:386``. + + ``revision_id`` is the UUID PK; ``rules_hash`` is hex-encoded so + it's canonically comparable to the same hash computed elsewhere + in the stack (K-6 Constitution pre-screen middleware, TS-side + ``constitutionVersionHashFromRulesHash``). + """ + + revision_id: str + rules_hash: str # hex-encoded + + +SELECT_ACTIVE_CONSTITUTION_REVISION_SQL = """ + SELECT + revision_id::text AS revision_id, + rules_hash AS rules_hash + FROM kronicle.workspace_constitution_revisions + WHERE workspace_id = $1 + ORDER BY revision_number DESC + LIMIT 1 +""" + + +def _hex_encode_rules_hash(value: Any) -> str: + """Normalize ``rules_hash`` from BYTEA / memoryview / str to hex. + + asyncpg returns BYTEA as ``bytes``; mocks in tests may return + a hex string already (pass-through). ``memoryview`` is the + asyncpg-bytea-buffer subtype on some versions. + """ + if isinstance(value, (bytes, bytearray)): + return value.hex() + if isinstance(value, memoryview): + return bytes(value).hex() + if isinstance(value, str): + return value + raise TypeError( + f"[kora.isokron] rules_hash arrived as unexpected type " + f"{type(value).__name__}; expected bytes / memoryview / str." + ) + + +async def read_active_constitution_revision( + workspace_id: str, + pool: Any, +) -> Optional[ActiveConstitutionRevision]: + """Return the active Constitution revision, or ``None`` for fresh workspaces. + + Fresh-workspace bootstrap state: a workspace that has not yet + authored any Constitution revisions has zero rows in this table. + Returning ``None`` (vs raising) matches the TS-side behavior; + callers either fall back to the platform-default Constitution + (PLAT-* rules in ``config/kronicle/default_constitution.yaml``) or + skip Constitution-dependent pre-screening for that turn. + + The RLS policy keys off ``app.current_workspace_id`` — same GUC + pattern as ``kora_policy_registry`` and ``agent_scratchpad_entries``. + Without the GUC set, the query returns zero rows silently; + set it in a transaction before the SELECT. + """ + async with pool.acquire() as conn: + async with conn.transaction(): + await conn.execute( + "SELECT set_config('app.current_workspace_id', $1, true)", + workspace_id, + ) + row = await conn.fetchrow( + SELECT_ACTIVE_CONSTITUTION_REVISION_SQL, workspace_id + ) + if row is None: + return None + return ActiveConstitutionRevision( + revision_id=row["revision_id"], + rules_hash=_hex_encode_rules_hash(row["rules_hash"]), + ) diff --git a/plugins/memory/isokron/events.py b/plugins/memory/isokron/events.py new file mode 100644 index 000000000000..3a604b625b86 --- /dev/null +++ b/plugins/memory/isokron/events.py @@ -0,0 +1,192 @@ +"""Chain event emit + recent events read (KR-2 ST4). + +Two halves: + +- :func:`read_recent_kora_events` — direct asyncpg read against + ``hivex_foundation.event_log`` filtered to ``event_type LIKE 'kora.%'``. + ``event_log`` is **tenant_id UUID keyed** (the one genuine substrate + exception to the workspace_id TEXT pattern, per foundation/0003); the + query resolves the caller's ``workspace_id`` (Clerk ``org_*`` TEXT) + to ``tenant_id`` via ``JOIN hivex_foundation.tenant ON + t.clerk_org_id = $1``. Matches the TS reader at + ``packages/sea-mcp-server/src/kora/context-assembler/index.ts:287``. + +- :func:`emit_kora_event` — deferred write surface. Chain events go + through the substrate's ``_emit_chain_event`` SECDEF (which sets + ``prev_event_hash`` / ``this_event_hash`` to maintain chain witness + integrity); calling it from runtime Python without the SECDEF wrapper + would break the witness chain. The path is a Sea MCP tool + (working name ``kora__append_event``); as of substrate main + ``28ff4f78``, no such tool is registered. Raises + :class:`ChainEventEmitNotAvailableError` until the tool ships. + BUILD_DEVIATIONS ``D-kr2-st4-no-chain-emit-mcp-tool``. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass +from typing import Any, Optional + +logger = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# Errors +# --------------------------------------------------------------------------- + + +class ChainEventEmitNotAvailableError(RuntimeError): + """Raised by :func:`emit_kora_event` until the Sea MCP tool ships. + + Same pattern as :class:`scratchpad.ScratchpadWriteNotAvailableError` + — runtime callers MUST NOT bypass with direct INSERT or + ``_emit_chain_event`` calls (would break chain witness integrity). + """ + + DEFAULT_MESSAGE = ( + "[kora.isokron.todo] chain event emit deferred — Sea MCP server " + "does not yet expose kora__append_event (or equivalent). Tracked " + "in BUILD_DEVIATIONS.md as D-kr2-st4-no-chain-emit-mcp-tool. " + "Direct INSERT into hivex_foundation.event_log bypasses the " + "prev_event_hash / this_event_hash chain — do NOT do that; the " + "MCP tool wraps the substrate's _emit_chain_event SECDEF which " + "preserves chain witness integrity." + ) + + def __init__(self, message: Optional[str] = None): + super().__init__(message or self.DEFAULT_MESSAGE) + + +# --------------------------------------------------------------------------- +# Typed shape +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True, slots=True) +class RecentChainEvent: + """One ``kora.*`` chain event from ``hivex_foundation.event_log``. + + Mirrors TS-side ``KoraRecentChainEvent`` in + ``packages/sea-mcp-server/src/kora/context-assembler/types.ts:72``. + ``payload_summary`` is ``jsonb_pretty(payload)`` truncated to 300 + chars + ellipsis — keeps system prompt bullets readable without + bloating the prompt on operator-extended payloads. + """ + + event_id: str + event_type: str + occurred_at: str # ISO-8601 from TIMESTAMPTZ + payload_summary: str + + +# --------------------------------------------------------------------------- +# Read SQL — matches TS reader verbatim +# --------------------------------------------------------------------------- + +SELECT_RECENT_KORA_CHAIN_EVENTS_SQL = """ + SELECT + el.event_id::text AS event_id, + el.event_type AS event_type, + el.occurred_at AS occurred_at, + jsonb_pretty(el.payload)::text AS payload_text + FROM hivex_foundation.event_log el + JOIN hivex_foundation.tenant t ON t.tenant_id = el.tenant_id + WHERE t.clerk_org_id = $1 + AND el.event_type LIKE 'kora.%' + ORDER BY el.occurred_at DESC + LIMIT $2 +""" + +DEFAULT_RECENT_EVENT_LIMIT = 50 +"""Per spec § ST4 + TS-side ``DEFAULT_RECENT_EVENT_LIMIT``.""" + +RECENT_EVENT_PAYLOAD_TRUNCATE_LENGTH = 300 +"""Per TS-side ``RECENT_EVENT_PAYLOAD_TRUNCATE_LENGTH``. Truncated +``jsonb_pretty`` output preserves the leading structural lines so an +operator skimming the system prompt can still see what each event +carried, without bloating long payloads.""" + + +# --------------------------------------------------------------------------- +# Read path +# --------------------------------------------------------------------------- + + +def _truncate_payload(payload_text: str) -> str: + if len(payload_text) <= RECENT_EVENT_PAYLOAD_TRUNCATE_LENGTH: + return payload_text + return payload_text[:RECENT_EVENT_PAYLOAD_TRUNCATE_LENGTH] + "…" + + +def _project_event(row: dict[str, Any]) -> RecentChainEvent: + occurred_at = row["occurred_at"] + if hasattr(occurred_at, "isoformat"): + occurred_at_iso = occurred_at.isoformat() + else: + occurred_at_iso = str(occurred_at) + return RecentChainEvent( + event_id=row["event_id"], + event_type=row["event_type"], + occurred_at=occurred_at_iso, + payload_summary=_truncate_payload(row["payload_text"] or ""), + ) + + +async def read_recent_kora_events( + workspace_id: str, + pool: Any, + *, + limit: int = DEFAULT_RECENT_EVENT_LIMIT, +) -> list[RecentChainEvent]: + """Read recent ``kora.*`` chain events for the workspace. + + Note: ``hivex_foundation.event_log`` keys off ``tenant_id UUID``, + not ``workspace_id TEXT`` like the other Kora tables. The JOIN + resolves the workspace_id (Clerk ``org_*``) to tenant_id via + ``hivex_foundation.tenant.clerk_org_id``. The TS reader at + ``packages/sea-mcp-server/src/kora/context-assembler/index.ts:287`` + uses the same JOIN; this is the substrate-team's established + resolution pattern. + + No RLS GUC needed here: ``event_log``'s row-security model uses + different machinery (per-partition partial indexes + foreign-key + constraints to tenant); the JOIN itself enforces tenant isolation + via the workspace_id binding. + """ + async with pool.acquire() as conn: + rows = await conn.fetch( + SELECT_RECENT_KORA_CHAIN_EVENTS_SQL, workspace_id, limit + ) + return [_project_event(dict(row)) for row in rows] + + +# --------------------------------------------------------------------------- +# Emit path (deferred — see BUILD_DEVIATIONS D-kr2-st4-no-chain-emit-mcp-tool) +# --------------------------------------------------------------------------- + + +async def emit_kora_event( + *, + workspace_id: str, + event_type: str, + payload: Any, + mcp_client: Any = None, +) -> str: + """Emit a ``kora.*`` chain event via the Sea MCP tool surface. + + Raises ``ChainEventEmitNotAvailableError`` until + ``kora__append_event`` (or equivalent) lands in the Sea MCP + server. Caller signature matches the future MCP-backed + implementation; when the tool ships the body switches to an + ``mcp_client.invoke('kora__append_event', ...)`` call without + any caller-side refactor. + + ``event_type`` must start with ``kora.`` and appear in the + ``event_log_event_type_check`` constraint set (foundation/0136 + + foundation/0138 ship the canonical vocabulary). Validation is + enforced substrate-side by the MCP tool — runtime callers pass + the literal through. + """ + del workspace_id, event_type, payload, mcp_client + raise ChainEventEmitNotAvailableError() diff --git a/plugins/memory/isokron/provider.py b/plugins/memory/isokron/provider.py index eef0a90b876a..9b63626524bc 100644 --- a/plugins/memory/isokron/provider.py +++ b/plugins/memory/isokron/provider.py @@ -1,23 +1,33 @@ """IsoKronMemoryProvider — Kora's substrate-backed MemoryProvider. -KR-2 ST1 shipped this as a structural skeleton. KR-2 ST2 wires the -read paths: +KR-2 closes here at ST4: every ABC method has a real implementation, +no ``NotImplementedError`` stubs remain. Surface summary: * **Role Charter** — ``read_active_role_charter`` against - ``public.kora_role_charter`` with SHA-256 integrity check. + ``public.kora_role_charter`` with SHA-256 integrity check (ST2). * **Capability matrix Kora row** — C2 Python mirror of ``ACTOR_CAPABILITY_MATRIX`` (parity test guards drift; K-7 will swap - to a Sea MCP tool). + to a Sea MCP tool; ST2). * **Policy registry** — ``read_kora_policy_registry`` against - ``kora_policy_registry`` with RLS GUC set inside a transaction; - 31-row sanity warned-on-drift. - -Subsequent sub-tasks fill in: - -* **ST3** — scratchpad reads + writes (Plan 02 schema). Wires - ``sync_turn`` / ``on_memory_write``. -* **ST4** — chain event emission (``kora.*`` event types) + recent - events read. Removes the remaining stubs. + ``kora_policy_registry`` with RLS GUC; 31-row sanity warn-on-drift (ST2). +* **Scratchpad** — own + cross-agent reads against + ``kronicle.agent_scratchpad_entries`` (ST3). Writes deferred behind + ``ScratchpadWriteNotAvailableError`` until K-8 ships the Sea MCP tool. +* **Recent chain events** — ``read_recent_kora_events`` against + ``hivex_foundation.event_log`` (tenant_id-keyed; JOIN tenant on + clerk_org_id) filtered ``LIKE 'kora.%'`` (ST4). +* **Active Constitution revision** — ``read_active_constitution_revision`` + against ``kronicle.workspace_constitution_revisions``; hex-encoded + ``rules_hash`` for K-6 Constitution pre-screen middleware (ST4). +* **Chain event emit** — deferred behind ``ChainEventEmitNotAvailableError`` + until Sea MCP ships ``kora__append_event`` (BUILD_DEVIATIONS + ``D-kr2-st4-no-chain-emit-mcp-tool``). Same shape as the scratchpad + write defer: catch the error in lifecycle hooks, log, continue. +* **Session context** — ``assemble_session_context`` returns a + ``KoraSessionContext`` mirroring the TS-side + ``packages/sea-mcp-server/src/kora/context-assembler/types.ts:130`` + shape; six load-bearing reads + two identity fields, fanned out via + ``asyncio.gather``. Selected via ``memory.provider: isokron`` in ``~/.kora/config.yaml``. Replaces Hermes' flat MEMORY.md / USER.md once configured; ``MemoryManager`` @@ -36,6 +46,11 @@ from .cache import TTLCache from .config import ISOKRON_CONFIG_SCHEMA, IsoKronProviderConfig from .connection import IsoKronConnection +from .events import ( + ChainEventEmitNotAvailableError, + RecentChainEvent, + emit_kora_event, +) from .models import ( KoraCapabilityRow, PolicyRegistryEntry, @@ -56,6 +71,7 @@ read_own_scratchpad, write_scratchpad_entry, ) +from .session_context import KoraSessionContext, assemble_session_context logger = logging.getLogger(__name__) @@ -84,22 +100,6 @@ ) -def _not_yet_implemented(method: str, sub_task: str) -> NotImplementedError: - """Build a Rule-6 honest NotImplementedError for ST1 skeleton stubs. - - Every stub method shares the same message format so operators can - grep `kora.isokron.todo` in their logs to see which surfaces are - still unimplemented. - """ - msg = ( - f"[kora.isokron.todo] IsoKronMemoryProvider.{method} not yet wired — " - f"KR-2 {sub_task} implements this. Rule-6: KR-2 ST1 shipped a " - f"structural skeleton; do not rely on this surface yet." - ) - logger.warning(msg) - return NotImplementedError(msg) - - class IsoKronMemoryProvider(MemoryProvider): """Substrate-backed MemoryProvider for the Kora runtime. @@ -167,6 +167,17 @@ def __init__(self, config: Optional[Dict[str, Any]] = None): self._cross_agent_scratchpad_cache: TTLCache[List[ScratchpadEntry]] = TTLCache( ttl_seconds=ttl ) + # Recent chain events + active Constitution revision (ST4 reads). + self._events_cache: TTLCache[List[RecentChainEvent]] = TTLCache( + ttl_seconds=ttl + ) + # Constitution revision cache holds Optional[tuple[str, str]] — + # (revision_id, rules_hash_hex) — or the sentinel ``(None, None)`` + # for fresh workspaces with no revisions. Using a tuple keeps the + # TTLCache invariant (cached value cannot be None for "miss"). + self._constitution_cache: TTLCache[ + tuple[Optional[str], Optional[str]] + ] = TTLCache(ttl_seconds=ttl) # Validate config eagerly so a typo surfaces at construct time # rather than at first turn. ``is_available`` checks this. @@ -278,13 +289,17 @@ def _resolve_workspace_id(self, **kwargs: Any) -> Optional[str]: return None def _prefetch_all(self, workspace_id: str) -> None: - """Block on a parallel ``asyncio.gather`` of the three reads. + """Block on a parallel ``asyncio.gather`` of all session reads. Each result populates its TTL cache so the subsequent ``system_prompt_block`` call hits warm cache. Integrity errors (RoleCharterIntegrityError, NoActiveRoleCharterError) surface as exceptions per spec § "fail-closed"; the policy 31-row - sanity warning is non-fatal. + sanity + scratchpad BLAKE3 drift are non-fatal WARNINGs. + + Seven reads in parallel: Role Charter, policy registry, + capability matrix, own scratchpad, cross-agent scratchpad, + recent ``kora.*`` chain events, active Constitution revision. Called by ``on_turn_start`` (per spec acceptance) and as a cache-warm step from ``system_prompt_block`` when the cache @@ -296,21 +311,98 @@ def _prefetch_all(self, workspace_id: str) -> None: ) pool = self._connection.get_pg_pool() - async def _gather() -> tuple[ - RoleCharter, List[PolicyRegistryEntry], KoraCapabilityRow - ]: + # We deliberately gather all 7 reads in one shot rather than + # call ``assemble_session_context`` — the latter doesn't fetch + # the policy registry (not part of KoraSessionContext), and + # we want a single gather for round-trip latency. + from .events import read_recent_kora_events + from .scratchpad import ( + read_cross_agent_scratchpad as _read_cross, + read_own_scratchpad as _read_own, + ) + from .constitution import read_active_constitution_revision + + async def _gather() -> Any: return await asyncio.gather( read_active_role_charter(workspace_id, pool), read_kora_policy_registry(workspace_id, pool), read_kora_capability_row(pool), + _read_own(workspace_id, pool), + _read_cross(workspace_id, pool), + read_recent_kora_events(workspace_id, pool), + read_active_constitution_revision(workspace_id, pool), ) - charter, policies, caps = self._connection.submit_and_wait( - _gather(), timeout=15.0 - ) + ( + charter, + policies, + caps, + own_entries, + cross_entries, + recent_events, + constitution, + ) = self._connection.submit_and_wait(_gather(), timeout=20.0) self._charter_cache.put(workspace_id, charter) self._policy_cache.put(workspace_id, policies) self._capability_cache.put(workspace_id, caps) + self._own_scratchpad_cache.put(workspace_id, own_entries) + self._cross_agent_scratchpad_cache.put(workspace_id, cross_entries) + self._events_cache.put(workspace_id, recent_events) + self._constitution_cache.put( + workspace_id, + ( + (constitution.revision_id, constitution.rules_hash) + if constitution is not None + else (None, None) + ), + ) + + def session_context( + self, *, workspace_id: Optional[str] = None + ) -> Optional[KoraSessionContext]: + """Return the assembled session context for the workspace. + + Reads from the post-prefetch caches; returns ``None`` if any + load-bearing cache is cold (call ``on_turn_start`` first to + warm). Mirrors the TS-side ``KoraSessionContext`` shape at + ``packages/sea-mcp-server/src/kora/context-assembler/types.ts:130``. + + Public API — consumers wanting just the typed context object + (e.g. K-6 Constitution pre-screen middleware in Python) call + this rather than touching individual caches. + """ + from datetime import datetime, timezone + + ws = self._resolve_workspace_id(workspace_id=workspace_id) + if ws is None: + return None + charter = self._charter_cache.get(ws) + capabilities = self._capability_cache.get(ws) + own = self._own_scratchpad_cache.get(ws) + cross = self._cross_agent_scratchpad_cache.get(ws) + events = self._events_cache.get(ws) + constitution = self._constitution_cache.get(ws) + if ( + charter is None + or capabilities is None + or own is None + or cross is None + or events is None + or constitution is None + ): + return None + rev_id, rules_hash = constitution + return KoraSessionContext( + workspace_id=ws, + assembled_at=datetime.now(timezone.utc).isoformat(), + role_charter=charter, + capability_matrix_row=capabilities, + own_scratchpad=tuple(own), + cross_agent_scratchpad=tuple(cross), + recent_chain_events=tuple(events), + active_constitution_revision_id=rev_id, + active_constitution_rules_hash=rules_hash, + ) def system_prompt_block(self) -> str: """Return the assembled identity prompt block. @@ -319,8 +411,9 @@ def system_prompt_block(self) -> str: §1 Identity — from ``content_md`` / sections.identity §2 CAN bullets — sections.authority_can_do §3 CANNOT bullets — sections.authority_cannot_do - Active policy values — selected 5 load-bearing rows - Capability matrix Kora-row summary — granted cap names + §4 Active policy values — selected 5 load-bearing rows + §5 Capability matrix Kora-row summary — granted cap names + §6 Recent ``kora.*`` activity — last few chain events Rule-6 honest-label — verbatim If the substrate is unreachable or the cache is cold and the @@ -344,19 +437,23 @@ def system_prompt_block(self) -> str: self._charter_cache.get(workspace_id) is None or self._policy_cache.get(workspace_id) is None or self._capability_cache.get(workspace_id) is None + or self._events_cache.get(workspace_id) is None ): self._prefetch_all(workspace_id) charter = self._charter_cache.get(workspace_id) policies = self._policy_cache.get(workspace_id) capabilities = self._capability_cache.get(workspace_id) - # All three are populated post-_prefetch_all; the ``is None`` - # guards are defensive (e.g. zero-TTL test config). + events = self._events_cache.get(workspace_id) or [] + # The four ST2/ST4 caches are populated post-_prefetch_all; the + # ``is None`` guards are defensive (e.g. zero-TTL test config). assert charter is not None, "charter cache miss after prefetch" assert policies is not None, "policy cache miss after prefetch" assert capabilities is not None, "capability cache miss after prefetch" - return _assemble_system_prompt_block(charter, policies, capabilities) + return _assemble_system_prompt_block( + charter, policies, capabilities, events + ) def prefetch(self, query: str, *, session_id: str = "") -> str: """Recall context for the upcoming turn — no-op in ST2. @@ -416,7 +513,15 @@ def handle_tool_call( args: Dict[str, Any], **kwargs: Any, ) -> str: - raise _not_yet_implemented("handle_tool_call", "ST3 (writes) / KR-3 (iso_node_* tools)") + """Handle a tool call routed by name. + + The provider returns no tools from ``get_tool_schemas`` (the + ``iso_node_*`` / ``iso_link_*`` family lands in KR-3), so this + hook should never be invoked in normal operation. Inherit the + ABC's "provider X does not handle tool Y" error so a routing + bug surfaces with a clear actionable message. + """ + return super().handle_tool_call(tool_name, args, **kwargs) # -- Scratchpad reads (sync wrappers around the async reads) ----------- @@ -551,7 +656,28 @@ def on_turn_start( self._prefetch_all(workspace_id) def on_session_end(self, messages: List[Dict[str, Any]]) -> None: - raise _not_yet_implemented("on_session_end", "ST4") + """Emit ``kora.session.ended`` chain event with turn count. + + Through the deferred-emit path until K-9 (or equivalent Sea + MCP ``kora__append_event`` tool) lands. The event would carry + ``{session_id, turn_count, ended_at}``; the catch + log + pattern preserves session lifecycle reliability regardless. + """ + workspace_id = self._resolve_workspace_id() + if workspace_id is None: + logger.debug( + "[kora.isokron] on_session_end: no workspace_id — chain event skipped." + ) + return + self._attempt_chain_event_emit( + workspace_id=workspace_id, + event_type="kora.session.ended", + payload={ + "session_id": self._session_id or "", + "turn_count": len(messages), + }, + origin="on_session_end", + ) def on_session_switch( self, @@ -561,10 +687,43 @@ def on_session_switch( reset: bool = False, **kwargs: Any, ) -> None: - raise _not_yet_implemented("on_session_switch", "ST4") + """Update stashed session_id + invalidate caches on a hard reset. + + ``/resume`` / ``/branch`` / compression keep the logical + conversation alive — leave the caches as-is; new session_id + is the only state to rotate. + + ``/reset`` / ``/new`` (``reset=True``) starts a fresh + conversation. Flush the per-workspace caches so the next turn + re-reads the current substrate state rather than serving + stale entries from a different logical session. + """ + del parent_session_id, kwargs + self._session_id = new_session_id + if reset: + self._charter_cache.clear() + self._policy_cache.clear() + self._capability_cache.clear() + self._own_scratchpad_cache.clear() + self._cross_agent_scratchpad_cache.clear() + self._events_cache.clear() + self._constitution_cache.clear() + logger.info( + "[kora.isokron] session reset to %s — all caches flushed", + new_session_id, + ) def on_pre_compress(self, messages: List[Dict[str, Any]]) -> str: - raise _not_yet_implemented("on_pre_compress", "ST4") + """Provider-extracted insights to preserve through compression. + + The IsoKron substrate is the system of record for everything + that should survive compression (Role Charter, scratchpad, + chain events). Conversation-history compression doesn't need + an isokron-side contribution; substrate-side data is + re-fetched on next turn. + """ + del messages + return "" def on_delegation( self, @@ -574,7 +733,40 @@ def on_delegation( child_session_id: str = "", **kwargs: Any, ) -> None: - raise _not_yet_implemented("on_delegation", "ST4") + """Record a subagent delegation as scratchpad + chain event. + + Subagents (e.g. claude_pm / oracle / critic) emit on their own + substrate identity; parent Kora records the observation via a + ``reasoning_trail`` scratchpad entry (so her own context shows + what she handed off + what came back) and a + ``kora.handoff.to_claude_pm`` chain event. + + Both paths are catch-and-continue (deferred MCP tools). + """ + del kwargs + workspace_id = self._resolve_workspace_id() + if workspace_id is None: + return + summary = _summarize_for_scratchpad( + f"[delegation child={child_session_id}] task={task!r} result={result!r}" + ) + self._attempt_scratchpad_write( + workspace_id=workspace_id, + scratchpad_kind=ScratchpadKind.REASONING_TRAIL, + visibility_scope=VisibilityScope.AGENT_PRIVATE, + content=summary, + origin="on_delegation", + ) + self._attempt_chain_event_emit( + workspace_id=workspace_id, + event_type="kora.handoff.to_claude_pm", + payload={ + "child_session_id": child_session_id, + "task_preview": task[:200], + "result_preview": result[:200], + }, + origin="on_delegation", + ) def on_memory_write( self, @@ -613,7 +805,57 @@ def on_memory_write( ) def save_config(self, values: Dict[str, Any], hermes_home: str) -> None: - raise _not_yet_implemented("save_config", "ST3 (`kora memory setup` walkthrough)") + """No-op: the IsoKron provider is configured via env vars + the + ``plugins.entries.isokron`` YAML block in ``config.yaml``. + + Per ABC: "Providers that use only env vars can leave the default + (no-op)". The ``kora memory setup`` walkthrough writes the YAML + block + an ``.env`` entry directly via the secret-handling path; + the provider itself has no native config file to maintain. + """ + del values, hermes_home + + # -- Chain event emit (deferred until Sea MCP tool ships) ---------------- + + def _attempt_chain_event_emit( + self, + *, + workspace_id: str, + event_type: str, + payload: Dict[str, Any], + origin: str, + ) -> None: + """Submit an emit via the dedicated IO loop; catch the defer-error. + + Mirrors :meth:`_attempt_scratchpad_write`'s defer-and-log + pattern. When the Sea MCP tool ships, only + :func:`events.emit_kora_event`'s body changes; this helper + stays identical. + """ + if self._connection is None: + raise RuntimeError( + f"[kora.isokron] _attempt_chain_event_emit ({origin}) before construct" + ) + try: + self._connection.submit_and_wait( + emit_kora_event( + workspace_id=workspace_id, + event_type=event_type, + payload=payload, + mcp_client=None, + ), + timeout=10.0, + ) + # Successful emit invalidates the events cache so the next + # system_prompt_block re-reads. + self._events_cache.invalidate(workspace_id) + except ChainEventEmitNotAvailableError as exc: + logger.warning( + "[kora.isokron] %s chain event emit skipped (%s) — %s", + origin, + event_type, + exc, + ) # --------------------------------------------------------------------------- @@ -682,8 +924,9 @@ def _assemble_system_prompt_block( charter: RoleCharter, policies: List[PolicyRegistryEntry], capabilities: KoraCapabilityRow, + recent_events: List[RecentChainEvent], ) -> str: - """Render the identity + policy + capability block for the system prompt. + """Render the identity + policy + capability + activity block. Pure function — extracted from ``IsoKronMemoryProvider`` so tests can drive it with synthetic shapes without mocking the connection. @@ -728,6 +971,40 @@ def _assemble_system_prompt_block( f"{len(capabilities.granted) + len(capabilities.denied)}):", cap_bullets, "", + _render_recent_activity(recent_events), + "", RULE_6_HONEST_LABEL, ] return "\n".join(blocks) + + +# Number of recent events to surface in the §6 prompt section. Keeps +# prompt size bounded — the full set lives in event_log and is +# queryable via read_recent_kora_events directly. +_SYSTEM_PROMPT_RECENT_EVENT_LIMIT = 5 + + +def _render_recent_activity(events: List[RecentChainEvent]) -> str: + """Render §6 'Recent kora.* activity'. + + Returns a "§6" section even when the list is empty so the section + structure stays consistent across sessions (operators inspecting + the prompt see the same anchors regardless of activity volume). + """ + if not events: + return "§6 Recent kora.* activity\n - " + head = events[:_SYSTEM_PROMPT_RECENT_EVENT_LIMIT] + bullets = [] + for evt in head: + # First line of payload_summary gives operators the shape of + # the event without the full pretty-printed JSON in-prompt. + first_line = evt.payload_summary.splitlines()[0] if evt.payload_summary else "" + bullets.append( + f" - {evt.occurred_at} {evt.event_type} {first_line}".rstrip() + ) + if len(events) > _SYSTEM_PROMPT_RECENT_EVENT_LIMIT: + bullets.append( + f" - …{len(events) - _SYSTEM_PROMPT_RECENT_EVENT_LIMIT} older " + f"event(s) in event_log" + ) + return f"§6 Recent kora.* activity ({len(events)} cached):\n" + "\n".join(bullets) diff --git a/plugins/memory/isokron/session_context.py b/plugins/memory/isokron/session_context.py new file mode 100644 index 000000000000..835b5aee2297 --- /dev/null +++ b/plugins/memory/isokron/session_context.py @@ -0,0 +1,111 @@ +"""KoraSessionContext shape + assembler (KR-2 ST4). + +Python mirror of TS-side ``KoraSessionContext`` at +``packages/sea-mcp-server/src/kora/context-assembler/types.ts:130``. +The session context is the single source of truth Kora reads at +session start; the six load-bearing fields are: + +1. ``role_charter`` — K-1 ST4 reader (ST2 here) +2. ``capability_matrix_row`` — Plan 04 ACTOR_CAPABILITY_MATRIX (ST2) +3. ``own_scratchpad`` — Plan 02 own entries (ST3) +4. ``cross_agent_scratchpad`` — Plan 02 cross-agent entries (ST3) +5. ``recent_chain_events`` — event_log filtered to ``kora.*`` (ST4) +6. ``active_constitution_revision_id`` + ``active_constitution_rules_hash`` + — Constitution revision (ST4) + +Plus two identity fields: +- ``workspace_id`` +- ``assembled_at`` — ISO-8601 timestamp captured at end of assembly + +The assembler fans out the six reads via ``asyncio.gather`` so cold +start hits one round-trip latency. Constitution-revision-absent +(fresh workspace) is non-fatal; ``active_constitution_revision_id`` / +``active_constitution_rules_hash`` are ``None`` in that case. +""" + +from __future__ import annotations + +import asyncio +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Any, Optional + +from .constitution import read_active_constitution_revision +from .events import RecentChainEvent, read_recent_kora_events +from .models import KoraCapabilityRow, RoleCharter +from .reads import read_active_role_charter, read_kora_capability_row +from .scratchpad import ( + ScratchpadEntry, + read_cross_agent_scratchpad, + read_own_scratchpad, +) + + +@dataclass(frozen=True, slots=True) +class KoraSessionContext: + """Six load-bearing reads + two identity fields, assembled once per turn.""" + + workspace_id: str + assembled_at: str # ISO-8601 + + role_charter: RoleCharter + capability_matrix_row: KoraCapabilityRow + own_scratchpad: tuple[ScratchpadEntry, ...] + cross_agent_scratchpad: tuple[ScratchpadEntry, ...] + recent_chain_events: tuple[RecentChainEvent, ...] + + # Constitution revision — None for fresh workspaces (no rows in + # kronicle.workspace_constitution_revisions yet). + active_constitution_revision_id: Optional[str] + active_constitution_rules_hash: Optional[str] + + +DEFAULT_SCRATCHPAD_LIMIT = 100 +DEFAULT_RECENT_EVENT_LIMIT = 50 + + +async def assemble_session_context( + workspace_id: str, + pool: Any, + *, + scratchpad_limit: int = DEFAULT_SCRATCHPAD_LIMIT, + recent_event_limit: int = DEFAULT_RECENT_EVENT_LIMIT, +) -> KoraSessionContext: + """Fan out the six reads in parallel via ``asyncio.gather``. + + Integrity errors from the Role Charter (SHA-256 mismatch / NULL + body) propagate — fail-closed per ST2 contract. Scratchpad + BLAKE3 drift warns but doesn't propagate. Constitution-revision- + absent is non-fatal (returns None pair). + """ + ( + role_charter, + capability_matrix_row, + own, + cross, + recent_events, + constitution, + ) = await asyncio.gather( + read_active_role_charter(workspace_id, pool), + read_kora_capability_row(pool), + read_own_scratchpad(workspace_id, pool, limit=scratchpad_limit), + read_cross_agent_scratchpad(workspace_id, pool, limit=scratchpad_limit), + read_recent_kora_events(workspace_id, pool, limit=recent_event_limit), + read_active_constitution_revision(workspace_id, pool), + ) + + return KoraSessionContext( + workspace_id=workspace_id, + assembled_at=datetime.now(timezone.utc).isoformat(), + role_charter=role_charter, + capability_matrix_row=capability_matrix_row, + own_scratchpad=tuple(own), + cross_agent_scratchpad=tuple(cross), + recent_chain_events=tuple(recent_events), + active_constitution_revision_id=( + constitution.revision_id if constitution is not None else None + ), + active_constitution_rules_hash=( + constitution.rules_hash if constitution is not None else None + ), + ) diff --git a/tests/plugins/memory/test_capability_matrix_parity.py b/tests/plugins/memory/test_capability_matrix_parity.py index 96ba14e5fb58..85881e7f7288 100644 --- a/tests/plugins/memory/test_capability_matrix_parity.py +++ b/tests/plugins/memory/test_capability_matrix_parity.py @@ -199,10 +199,10 @@ def test_python_mirror_matches_ts_source_for_every_kora_column_value( def test_python_mirror_has_expected_subset_counts(): - """Sanity: 24 SEA + 24 KORA_BROADER = 48 in the combined dict.""" + """24 SEA + 25 KORA_BROADER = 49 (post-cap_unbless_convention add).""" assert len(SEA_CAPABILITIES_KORA_COLUMN) == 24 - assert len(KORA_BROADER_CAPABILITIES_KORA_COLUMN) == 24 - assert len(ACTOR_CAPABILITY_MATRIX_KORA_COLUMN) == 48 + assert len(KORA_BROADER_CAPABILITIES_KORA_COLUMN) == 25 + assert len(ACTOR_CAPABILITY_MATRIX_KORA_COLUMN) == 49 # No overlap between subsets. overlap = set(SEA_CAPABILITIES_KORA_COLUMN) & set( KORA_BROADER_CAPABILITIES_KORA_COLUMN diff --git a/tests/plugins/memory/test_constitution.py b/tests/plugins/memory/test_constitution.py new file mode 100644 index 000000000000..930c7128ca37 --- /dev/null +++ b/tests/plugins/memory/test_constitution.py @@ -0,0 +1,157 @@ +"""KR-2 ST4 — Active Constitution revision read.""" + +from __future__ import annotations + +import asyncio +from typing import Any, Optional + +import pytest + +from plugins.memory.isokron.constitution import ( + ActiveConstitutionRevision, + SELECT_ACTIVE_CONSTITUTION_REVISION_SQL, + _hex_encode_rules_hash, + read_active_constitution_revision, +) + + +# --------------------------------------------------------------------------- +# Fake pool / conn (records call order for the RLS-GUC ordering check) +# --------------------------------------------------------------------------- + + +class _FakeTxn: + def __init__(self, conn): + self._conn = conn + + async def __aenter__(self): + self._conn.calls.append(("txn.enter",)) + return self + + async def __aexit__(self, *exc): + self._conn.calls.append(("txn.exit",)) + return False + + +class _FakeConnection: + def __init__(self, fetchrow_result: Optional[dict[str, Any]] = None): + self.fetchrow_result = fetchrow_result + self.calls: list[tuple] = [] + + async def execute(self, sql: str, *args): + self.calls.append(("execute", sql, args)) + return "OK" + + async def fetchrow(self, sql: str, *args): + self.calls.append(("fetchrow", sql, args)) + return self.fetchrow_result + + def transaction(self): + return _FakeTxn(self) + + +class _FakeAcquireCtx: + def __init__(self, conn): + self._conn = conn + + async def __aenter__(self): + return self._conn + + async def __aexit__(self, *exc): + return False + + +class _FakePool: + def __init__(self, conn): + self._conn = conn + + def acquire(self): + return _FakeAcquireCtx(self._conn) + + +WORKSPACE_ID = "org_test_workspace_001" + + +# --------------------------------------------------------------------------- +# Happy path +# --------------------------------------------------------------------------- + + +def test_read_active_constitution_revision_returns_typed_shape(): + row = { + "revision_id": "11111111-1111-1111-1111-111111111111", + "rules_hash": b"\xde\xad\xbe\xef" + b"\x00" * 28, # 32 bytes (BLAKE3/SHA-256 size) + } + conn = _FakeConnection(fetchrow_result=row) + pool = _FakePool(conn) + rev = asyncio.run(read_active_constitution_revision(WORKSPACE_ID, pool)) + assert isinstance(rev, ActiveConstitutionRevision) + assert rev.revision_id == "11111111-1111-1111-1111-111111111111" + assert rev.rules_hash == "deadbeef" + "00" * 28 + + +def test_read_active_constitution_revision_returns_none_for_fresh_workspace(): + """Workspace with no Constitution revisions → returns ``None`` (not raises).""" + conn = _FakeConnection(fetchrow_result=None) + pool = _FakePool(conn) + rev = asyncio.run(read_active_constitution_revision(WORKSPACE_ID, pool)) + assert rev is None + + +def test_read_active_constitution_revision_sets_rls_guc_before_fetch(): + """Required call sequence: txn.enter → execute(set_config) → fetchrow → txn.exit.""" + conn = _FakeConnection(fetchrow_result=None) + pool = _FakePool(conn) + asyncio.run(read_active_constitution_revision(WORKSPACE_ID, pool)) + methods = [c[0] for c in conn.calls] + assert methods == [ + "txn.enter", + "execute", + "fetchrow", + "txn.exit", + ], f"call sequence: {conn.calls}" + _, exec_sql, exec_args = conn.calls[1] + assert "set_config" in exec_sql + assert "app.current_workspace_id" in exec_sql + assert exec_args == (WORKSPACE_ID,) + + +def test_read_active_constitution_revision_sql_orders_by_revision_number_desc(): + """No superseded_at; active = highest revision_number. + + The TS-side reader caught the K-DG drift (earlier bucket prompt + referenced WHERE superseded_at IS NULL which would fail). Verify + the corrected SQL ships here. + """ + sql = SELECT_ACTIVE_CONSTITUTION_REVISION_SQL + assert "ORDER BY revision_number DESC" in sql + assert "LIMIT 1" in sql + assert "superseded_at" not in sql # must NOT use the nonexistent column + + +# --------------------------------------------------------------------------- +# Hex encoder +# --------------------------------------------------------------------------- + + +def test_hex_encode_accepts_bytes(): + assert _hex_encode_rules_hash(b"\xab\xcd") == "abcd" + + +def test_hex_encode_accepts_bytearray(): + assert _hex_encode_rules_hash(bytearray(b"\xff\x00")) == "ff00" + + +def test_hex_encode_accepts_memoryview(): + assert _hex_encode_rules_hash(memoryview(b"\x12\x34")) == "1234" + + +def test_hex_encode_passes_through_string(): + """Tests with mocks may pass a hex string directly — accept it.""" + assert _hex_encode_rules_hash("already_hex") == "already_hex" + + +def test_hex_encode_rejects_unknown_type(): + with pytest.raises(TypeError) as excinfo: + _hex_encode_rules_hash(42) + assert "rules_hash arrived as unexpected type" in str(excinfo.value) diff --git a/tests/plugins/memory/test_events.py b/tests/plugins/memory/test_events.py new file mode 100644 index 000000000000..acb99776ed7b --- /dev/null +++ b/tests/plugins/memory/test_events.py @@ -0,0 +1,186 @@ +"""KR-2 ST4 — Chain event emit + recent events read.""" + +from __future__ import annotations + +import asyncio +from typing import Any, List, Optional + +import pytest + +from plugins.memory.isokron.events import ( + ChainEventEmitNotAvailableError, + DEFAULT_RECENT_EVENT_LIMIT, + RECENT_EVENT_PAYLOAD_TRUNCATE_LENGTH, + RecentChainEvent, + SELECT_RECENT_KORA_CHAIN_EVENTS_SQL, + emit_kora_event, + read_recent_kora_events, +) + + +# --------------------------------------------------------------------------- +# Fake pool / connection +# --------------------------------------------------------------------------- + + +class _FakeConnection: + def __init__(self, rows: Optional[List[dict[str, Any]]] = None): + self.rows = rows or [] + self.calls: list[tuple] = [] + + async def fetch(self, sql: str, *args): + self.calls.append(("fetch", sql, args)) + return self.rows + + +class _FakeAcquireCtx: + def __init__(self, conn): + self._conn = conn + + async def __aenter__(self): + return self._conn + + async def __aexit__(self, *exc): + return False + + +class _FakePool: + def __init__(self, conn): + self._conn = conn + + def acquire(self): + return _FakeAcquireCtx(self._conn) + + +WORKSPACE_ID = "org_test_workspace_001" + + +def _event_row( + *, + event_id: str, + event_type: str = "kora.recommendation.issued", + occurred_at: Any = "2026-05-20T12:00:00Z", + payload_text: str = '{"foo":"bar"}', +) -> dict[str, Any]: + return { + "event_id": event_id, + "event_type": event_type, + "occurred_at": occurred_at, + "payload_text": payload_text, + } + + +# --------------------------------------------------------------------------- +# Read recent events +# --------------------------------------------------------------------------- + + +def test_read_recent_kora_events_returns_typed_entries(): + rows = [ + _event_row(event_id="a", event_type="kora.recommendation.issued"), + _event_row(event_id="b", event_type="kora.escalation.requested"), + ] + conn = _FakeConnection(rows) + pool = _FakePool(conn) + events = asyncio.run(read_recent_kora_events(WORKSPACE_ID, pool)) + assert len(events) == 2 + assert all(isinstance(e, RecentChainEvent) for e in events) + assert events[0].event_type == "kora.recommendation.issued" + assert events[1].event_type == "kora.escalation.requested" + + +def test_read_recent_kora_events_binds_workspace_id_and_default_limit(): + """SQL is bound with (workspace_id, limit); default limit is 50.""" + conn = _FakeConnection([]) + pool = _FakePool(conn) + asyncio.run(read_recent_kora_events(WORKSPACE_ID, pool)) + _, _sql, args = conn.calls[0] + assert args == (WORKSPACE_ID, 50) + assert DEFAULT_RECENT_EVENT_LIMIT == 50 + + +def test_read_recent_kora_events_passes_custom_limit(): + conn = _FakeConnection([]) + pool = _FakePool(conn) + asyncio.run(read_recent_kora_events(WORKSPACE_ID, pool, limit=10)) + _, _sql, args = conn.calls[0] + assert args == (WORKSPACE_ID, 10) + + +def test_read_recent_kora_events_sql_joins_tenant_on_clerk_org_id(): + """The SQL must JOIN hivex_foundation.tenant on clerk_org_id. + + event_log is the substrate's one genuine tenant_id-UUID-keyed table; + the JOIN translates Kora's workspace_id (Clerk TEXT) into tenant_id. + Missing the JOIN would either fail (no FK match) or silently + return zero rows. + """ + sql = SELECT_RECENT_KORA_CHAIN_EVENTS_SQL + assert "JOIN hivex_foundation.tenant" in sql + assert "t.clerk_org_id = $1" in sql + assert "el.event_type LIKE 'kora.%'" in sql + assert "ORDER BY el.occurred_at DESC" in sql + + +def test_read_recent_kora_events_truncates_payload_at_300_chars(): + long_payload = "x" * 500 + rows = [_event_row(event_id="a", payload_text=long_payload)] + conn = _FakeConnection(rows) + pool = _FakePool(conn) + events = asyncio.run(read_recent_kora_events(WORKSPACE_ID, pool)) + assert len(events[0].payload_summary) == RECENT_EVENT_PAYLOAD_TRUNCATE_LENGTH + 1 + assert events[0].payload_summary.endswith("…") + # First 300 chars are the original text. + assert events[0].payload_summary[:300] == "x" * 300 + + +def test_read_recent_kora_events_short_payload_passes_through_unchanged(): + rows = [_event_row(event_id="a", payload_text="short")] + conn = _FakeConnection(rows) + pool = _FakePool(conn) + events = asyncio.run(read_recent_kora_events(WORKSPACE_ID, pool)) + assert events[0].payload_summary == "short" + + +def test_read_recent_kora_events_empty_payload_handled(): + rows = [_event_row(event_id="a", payload_text="")] + conn = _FakeConnection(rows) + pool = _FakePool(conn) + events = asyncio.run(read_recent_kora_events(WORKSPACE_ID, pool)) + assert events[0].payload_summary == "" + + +def test_read_recent_kora_events_handles_datetime_occurred_at(): + """If asyncpg returns occurred_at as datetime, it gets ISO-encoded.""" + from datetime import datetime, timezone + + rows = [ + _event_row(event_id="a", occurred_at=datetime(2026, 5, 20, 12, 0, tzinfo=timezone.utc)), + ] + conn = _FakeConnection(rows) + pool = _FakePool(conn) + events = asyncio.run(read_recent_kora_events(WORKSPACE_ID, pool)) + assert "2026-05-20T12:00:00" in events[0].occurred_at + + +# --------------------------------------------------------------------------- +# Deferred emit +# --------------------------------------------------------------------------- + + +def test_emit_kora_event_raises_deferred_write_error(): + """Until the Sea MCP tool ships, emit raises ChainEventEmitNotAvailableError.""" + + async def _run(): + await emit_kora_event( + workspace_id=WORKSPACE_ID, + event_type="kora.session.ended", + payload={"turn_count": 5}, + ) + + with pytest.raises(ChainEventEmitNotAvailableError) as excinfo: + asyncio.run(_run()) + msg = str(excinfo.value) + assert "[kora.isokron.todo]" in msg + assert "D-kr2-st4-no-chain-emit-mcp-tool" in msg + assert "BUILD_DEVIATIONS" in msg diff --git a/tests/plugins/memory/test_isokron_provider_skeleton.py b/tests/plugins/memory/test_isokron_provider_skeleton.py index 0ca3742f2e1b..ec59788e88f0 100644 --- a/tests/plugins/memory/test_isokron_provider_skeleton.py +++ b/tests/plugins/memory/test_isokron_provider_skeleton.py @@ -206,37 +206,43 @@ def test_env_var_expansion_missing_leaves_literal(monkeypatch): # --------------------------------------------------------------------------- -@pytest.mark.parametrize( - "method, args, kwargs", - [ - # ST2 implemented: system_prompt_block, on_turn_start (reads). - # ST2 no-op'd: prefetch, queue_prefetch (ABC defaults). - # ST3 implemented: sync_turn, on_memory_write (deferred-write - # surface; ScratchpadWriteNotAvailableError caught + logged). - # Remaining stubs target ST4 / KR-3. - ("handle_tool_call", ("t", {}), {}), - ("on_session_end", ([],), {}), - ("on_session_switch", ("new-id",), {"reset": True}), - ("on_pre_compress", ([],), {}), - ("on_delegation", ("task", "result"), {"child_session_id": "c"}), - ("save_config", ({"key": "val"}, "/tmp/kora-home"), {}), - ], -) -def test_stub_method_raises_with_rule6_message(method, args, kwargs): - """Each remaining stub raises NotImplementedError tagged ``[kora.isokron.todo]``.""" +def test_no_stub_methods_remain_after_st4(): + """KR-2 ST4 close-out: every ABC method has a real implementation. + + ``handle_tool_call`` inherits the ABC default which raises a clear + error when invoked with an unsupported tool name — that's the + correct behavior for a no-tools provider (``get_tool_schemas`` + returns ``[]``) and is verified separately in + ``test_handle_tool_call_unsupported_tool_raises_with_provider_name``. + All other ABC methods return normally. + """ + from plugins.memory.isokron.provider import IsoKronMemoryProvider + + provider = IsoKronMemoryProvider(config=_minimal_config()) + provider.initialize(session_id="t-st4-stubcheck") + try: + # None of these should raise NotImplementedError now. + assert provider.prefetch("q", session_id="s") == "" + assert provider.queue_prefetch("q", session_id="s") is None + assert provider.on_pre_compress([]) == "" + # save_config no-op: + assert provider.save_config({"k": "v"}, "/tmp/kora-home") is None + # Lifecycle hooks with no workspace_id resolution path will + # log debug and return — no exception: + provider.on_session_switch("new-sess", reset=False) + finally: + provider.shutdown() + + +def test_handle_tool_call_unsupported_tool_raises_with_provider_name(): + """handle_tool_call inherits the ABC default since provider has no tools.""" from plugins.memory.isokron.provider import IsoKronMemoryProvider provider = IsoKronMemoryProvider(config=_minimal_config()) - fn = getattr(provider, method) with pytest.raises(NotImplementedError) as excinfo: - fn(*args, **kwargs) - assert "[kora.isokron.todo]" in str(excinfo.value), ( - f"{method} missing Rule-6 todo tag: {excinfo.value}" - ) - msg = str(excinfo.value) - assert any(tag in msg for tag in ("ST3", "ST4", "KR-3")), ( - f"{method} stub message missing forward-target tag: {msg}" - ) + provider.handle_tool_call("some_tool_name_kr_3_will_add", {}) + assert "isokron" in str(excinfo.value) + assert "some_tool_name_kr_3_will_add" in str(excinfo.value) # --------------------------------------------------------------------------- diff --git a/tests/plugins/memory/test_provider_end_to_end.py b/tests/plugins/memory/test_provider_end_to_end.py new file mode 100644 index 000000000000..5da7975fde5c --- /dev/null +++ b/tests/plugins/memory/test_provider_end_to_end.py @@ -0,0 +1,314 @@ +"""KR-2 ST4 — End-to-end provider lifecycle test. + +Walks a synthesized session through the full IsoKronMemoryProvider +lifecycle, mocking only the connection layer (asyncpg pool + the +dedicated IO loop's ``submit_and_wait``). Verifies that no +``NotImplementedError`` surfaces anywhere and that integrity checks +all pass. + +Lifecycle covered: + + initialize(session_id) + on_turn_start(turn=1, message) # warms 7 reads + system_prompt_block() # uses warm cache + sync_turn(user, "...cap_X...") # attempts scratchpad write + on_memory_write("add", "memory", "...") # mirrors built-in writes + on_delegation(task, result, child_id) # emits + mirrors + on_session_end([...]) # emits kora.session.ended + shutdown() +""" + +from __future__ import annotations + +import asyncio +import hashlib +import logging +from typing import Any, List, Optional + +import pytest + + +# --------------------------------------------------------------------------- +# Synthetic substrate fixtures +# --------------------------------------------------------------------------- + + +WORKSPACE_ID = "org_e2e_workspace" + +_CONTENT_MD = "# Kora Role Charter v1.0\n\nKora is bounded operator-tier-plus." +_CONTENT_HASH = hashlib.sha256(_CONTENT_MD.encode("utf-8")).hexdigest() + +_JSONB = { + "schema_version": 1, + "charter_version": "1.0", + "sections": { + "identity": "Kora is bounded.", + "authority_can_do": ["Propose policy", "Write scratchpad"], + "authority_cannot_do": ["Override security verdicts"], + "override_preconditions": ["6-precondition firewall"], + "escalation_triggers": ["Novel class-1 decisions"], + "per_session_discipline": ["Pre-fetch on session start"], + "audit_attribution": "kora.* chain events", + "charter_modification": "operator-direct", + "effective_date_clause": "2026-05-20", + }, +} + + +def _charter_row(): + return { + "id": "11111111-1111-1111-1111-111111111111", + "workspace_id": WORKSPACE_ID, + "schema_version": 1, + "content_md": _CONTENT_MD, + "content_jsonb": _JSONB, + "content_hash": _CONTENT_HASH, + "created_at": "2026-05-20T00:00:00Z", + } + + +def _policy_rows(n: int = 31) -> list[dict[str, Any]]: + return [ + {"policy_path": f"policy.kora_synth_{i}", "policy_value": True} + for i in range(n) + ] + + +def _event_rows() -> list[dict[str, Any]]: + return [ + { + "event_id": "ee1", + "event_type": "kora.recommendation.issued", + "occurred_at": "2026-05-20T12:00:00Z", + "payload_text": '{"item":"watch_brief"}', + }, + { + "event_id": "ee2", + "event_type": "kora.handoff.to_claude_pm", + "occurred_at": "2026-05-20T12:05:00Z", + "payload_text": '{"child":"sess-x"}', + }, + ] + + +# --------------------------------------------------------------------------- +# Fake connection — routes each SQL by content keyword +# --------------------------------------------------------------------------- + + +class _FakeTxn: + def __init__(self, conn): + self._conn = conn + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + +class _FakeConnection: + def __init__(self): + self.calls: list[tuple] = [] + + async def execute(self, sql: str, *args): + self.calls.append(("execute", sql[:50], args)) + return "OK" + + async def fetchrow(self, sql: str, *args): + self.calls.append(("fetchrow", sql[:50], args)) + if "kora_role_charter" in sql: + return _charter_row() + if "workspace_constitution_revisions" in sql: + return { + "revision_id": "33333333-3333-3333-3333-333333333333", + "rules_hash": b"\xfe\xed" * 16, + } + return None + + async def fetch(self, sql: str, *args): + self.calls.append(("fetch", sql[:50], args)) + if "kora_policy_registry" in sql: + return _policy_rows(31) + if "ar.actor_kind = 'kora'" in sql: + return [] + if "ar.actor_kind != 'kora'" in sql: + return [] + if "LIKE 'kora.%'" in sql: + return _event_rows() + return [] + + def transaction(self): + return _FakeTxn(self) + + +class _FakeAcquireCtx: + def __init__(self, conn): + self._conn = conn + + async def __aenter__(self): + return self._conn + + async def __aexit__(self, *exc): + return False + + +class _FakePool: + def __init__(self, conn): + self._conn = conn + + def acquire(self): + return _FakeAcquireCtx(self._conn) + + +class _FakeProviderConnection: + """Drop-in for IsoKronConnection in the E2E test.""" + + def __init__(self): + self._conn = _FakeConnection() + self._pool = _FakePool(self._conn) + self.submitted: list = [] + self.closed = False + + def get_pg_pool(self): + return self._pool + + def submit_and_wait(self, coro, *, timeout: float = 10.0): + self.submitted.append(coro) + # Run the coroutine on a one-off loop so deferred-write + # errors propagate exactly as they would in production. + return asyncio.run(coro) + + def close(self): + self.closed = True + + # No-op shims so provider.shutdown() works without the real + # connection lifecycle. + @property + def is_started(self): + return not self.closed + + def start(self): # pragma: no cover — initialize calls but we don't need it + pass + + +# --------------------------------------------------------------------------- +# E2E test +# --------------------------------------------------------------------------- + + +def test_provider_end_to_end_full_lifecycle(caplog): + """Walks every ABC method without surfacing any NotImplementedError.""" + from plugins.memory.isokron.provider import IsoKronMemoryProvider + + provider = IsoKronMemoryProvider( + config={ + "isokron_dsn": "postgres://kora:secret@localhost:5432/isokron", + "mcp_endpoint": "stdio://node ./sea-mcp-server.js", + "default_workspace_id": WORKSPACE_ID, + } + ) + fake_conn = _FakeProviderConnection() + setattr(provider, "_connection", fake_conn) + + with caplog.at_level(logging.DEBUG, logger="plugins.memory.isokron"): + # 1. Initialize — stash session_id; IO loop is already "started" + # via fake connection's start(). + provider.initialize(session_id="e2e-001", platform="cli") + assert provider._initialized is True + + # 2. on_turn_start — warms 7 caches via the gather. + provider.on_turn_start(turn_number=1, message="user opens session") + + # All caches warm post-prefetch. + assert WORKSPACE_ID in provider._charter_cache + assert WORKSPACE_ID in provider._policy_cache + assert WORKSPACE_ID in provider._capability_cache + assert WORKSPACE_ID in provider._own_scratchpad_cache + assert WORKSPACE_ID in provider._cross_agent_scratchpad_cache + assert WORKSPACE_ID in provider._events_cache + assert WORKSPACE_ID in provider._constitution_cache + + # 3. system_prompt_block — uses warm cache; non-empty; all sections. + block = provider.system_prompt_block() + assert block + assert "§1 Identity" in block + assert "§2 You CAN:" in block + assert "§3 You CANNOT:" in block + assert "§4 Active policy values" in block + assert "§5 Granted capabilities" in block + assert "§6 Recent kora.* activity" in block + assert "kora.recommendation.issued" in block + assert "kora.handoff.to_claude_pm" in block + # Rule-6 honest-label verbatim: + assert "This identity block was assembled by IsoKronMemoryProvider" in block + + # 4. session_context — typed shape returned from warm caches. + ctx = provider.session_context() + assert ctx is not None + assert ctx.workspace_id == WORKSPACE_ID + assert ctx.active_constitution_revision_id == "33333333-3333-3333-3333-333333333333" + # rules_hash hex-encoded. + assert ctx.active_constitution_rules_hash == "feed" * 16 + assert len(ctx.recent_chain_events) == 2 + + # 5. sync_turn — Kora-action token in assistant → attempts write + # → catches the deferred-error → logs. + provider.sync_turn( + "what should I work on?", + "ok, calling cap_propose_policy_change now", + session_id="e2e-001", + ) + + # 6. on_memory_write — mirrors to scratchpad. + provider.on_memory_write( + "add", "user", "Joshua likes morning coffee." + ) + + # 7. on_delegation — emits + scratchpad mirror (both deferred but + # both attempted). + provider.on_delegation( + "compile a watch brief", + "watch brief item: x", + child_session_id="claude_pm-sess-001", + ) + + # 8. on_pre_compress — returns empty per design. + assert provider.on_pre_compress([{"role": "user", "content": "x"}]) == "" + + # 9. on_session_switch — soft (no reset) keeps caches. + provider.on_session_switch("e2e-002", reset=False) + assert WORKSPACE_ID in provider._charter_cache + + # 10. on_session_end — emits kora.session.ended (deferred). + provider.on_session_end([{"role": "user", "content": "bye"}]) + + # 11. on_session_switch with reset=True — flushes all caches. + provider.on_session_switch("e2e-003", reset=True) + assert WORKSPACE_ID not in provider._charter_cache + assert WORKSPACE_ID not in provider._events_cache + + # No NotImplementedError surfaced anywhere through the full lifecycle. + # Deferred-write WARNINGs were logged but caught. + deferred = [r for r in caplog.records if "skipped" in r.getMessage()] + # Expect at least: sync_turn (1) + on_memory_write (1) + on_delegation + # scratchpad (1) + on_delegation emit (1) + on_session_end emit (1) = 5 + assert len(deferred) >= 5 + + provider.shutdown() + assert provider._initialized is False + + +def test_provider_session_context_returns_none_when_cache_cold(): + """Before on_turn_start warms the caches, session_context returns None.""" + from plugins.memory.isokron.provider import IsoKronMemoryProvider + + provider = IsoKronMemoryProvider( + config={ + "isokron_dsn": "postgres://x@y/z", + "mcp_endpoint": "stdio://x", + "default_workspace_id": WORKSPACE_ID, + } + ) + # No connection set; no prefetch run. + assert provider.session_context() is None diff --git a/tests/plugins/memory/test_reads.py b/tests/plugins/memory/test_reads.py index 9d6452b9990d..cbf50fb74ab6 100644 --- a/tests/plugins/memory/test_reads.py +++ b/tests/plugins/memory/test_reads.py @@ -390,8 +390,8 @@ def test_read_kora_policy_registry_passes_codec_decoded_values_through(): def test_capability_mirror_loads_with_expected_shape(): """The mirror is a non-empty dict with the right keys + boolean values.""" assert isinstance(ACTOR_CAPABILITY_MATRIX_KORA_COLUMN, dict) - # 24 SEA_CAPABILITIES + 24 KORA_BROADER_CAPABILITIES = 48 entries. - assert len(ACTOR_CAPABILITY_MATRIX_KORA_COLUMN) == 48 + # 24 SEA + 25 KORA_BROADER = 49 entries (cap_unbless_convention added 2026-05-20). + assert len(ACTOR_CAPABILITY_MATRIX_KORA_COLUMN) == 49 # Values are booleans (not strings, not ints). assert all( isinstance(v, bool) for v in ACTOR_CAPABILITY_MATRIX_KORA_COLUMN.values() @@ -405,10 +405,11 @@ def test_read_kora_capability_row_returns_typed_kora_row(): row = asyncio.run(read_kora_capability_row()) assert isinstance(row, KoraCapabilityRow) assert row.actor_kind == "kora" - # Kora has 22 granted caps (3 sea + 19 kora-broader) of the 48 total. + # Kora has 22 granted caps (3 sea + 19 kora-broader) of the 49 total. + # cap_unbless_convention is operator-only → denied for Kora. assert len(row.granted) == 22 - assert len(row.denied) == 26 - assert len(row.granted) + len(row.denied) == 48 + assert len(row.denied) == 27 + assert len(row.granted) + len(row.denied) == 49 def test_kora_capability_row_has_lookup_is_fail_closed(): @@ -507,7 +508,10 @@ def _synthetic_capabilities() -> KoraCapabilityRow: def test_system_prompt_block_assembles_all_required_sections(): """The assembler produces non-empty text with each spec-required section.""" block = _assemble_system_prompt_block( - _synthetic_charter(), _synthetic_policies(), _synthetic_capabilities() + _synthetic_charter(), + _synthetic_policies(), + _synthetic_capabilities(), + [], # no recent events for this base-shape test ) assert block # non-empty # §1 Identity @@ -527,6 +531,8 @@ def test_system_prompt_block_assembles_all_required_sections(): # §5 granted capabilities (Kora has 22 granted of 48) assert "§5 Granted capabilities" in block assert "cap_write_agent_scratchpad" in block + # §6 recent activity — present even with empty list + assert "§6 Recent kora.* activity" in block # Granted caps are sorted — never references operator-only caps as granted assert "cap_override_security_or_policy_verdict" not in block.split( "§5 Granted capabilities" @@ -536,7 +542,10 @@ def test_system_prompt_block_assembles_all_required_sections(): def test_system_prompt_block_contains_rule_6_label_verbatim(): """Rule-6 honest-label appears verbatim — operators grep for this string.""" block = _assemble_system_prompt_block( - _synthetic_charter(), _synthetic_policies(), _synthetic_capabilities() + _synthetic_charter(), + _synthetic_policies(), + _synthetic_capabilities(), + [], ) assert RULE_6_HONEST_LABEL in block # And it's at the bottom (last line of the block). @@ -549,6 +558,7 @@ def test_system_prompt_block_marks_missing_policy_entries(): _synthetic_charter(), [], # empty registry — every entry shows _synthetic_capabilities(), + [], ) for path in SYSTEM_PROMPT_POLICY_PATHS: assert f"{path} = " in block diff --git a/tests/plugins/memory/test_session_context.py b/tests/plugins/memory/test_session_context.py new file mode 100644 index 000000000000..dd84a11f15b8 --- /dev/null +++ b/tests/plugins/memory/test_session_context.py @@ -0,0 +1,176 @@ +"""KR-2 ST4 — KoraSessionContext assembler.""" + +from __future__ import annotations + +import asyncio +import hashlib +from typing import Any, Optional + +import pytest + +from plugins.memory.isokron.session_context import ( + DEFAULT_RECENT_EVENT_LIMIT, + DEFAULT_SCRATCHPAD_LIMIT, + KoraSessionContext, + assemble_session_context, +) + + +# --------------------------------------------------------------------------- +# Combined fake — serves Role Charter / scratchpad / events / constitution +# all from one connection. +# --------------------------------------------------------------------------- + + +class _FakeTxn: + def __init__(self, conn): + self._conn = conn + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + +class _FakeConnection: + """Fake conn that routes queries based on SQL content keywords.""" + + def __init__( + self, + *, + role_charter_row: Optional[dict[str, Any]] = None, + own_scratchpad_rows: Optional[list[dict[str, Any]]] = None, + cross_agent_rows: Optional[list[dict[str, Any]]] = None, + recent_event_rows: Optional[list[dict[str, Any]]] = None, + constitution_row: Optional[dict[str, Any]] = None, + ): + self.role_charter_row = role_charter_row + self.own_scratchpad_rows = own_scratchpad_rows or [] + self.cross_agent_rows = cross_agent_rows or [] + self.recent_event_rows = recent_event_rows or [] + self.constitution_row = constitution_row + self.calls: list[tuple] = [] + + async def execute(self, sql: str, *args): + self.calls.append(("execute", sql[:60], args)) + return "OK" + + async def fetchrow(self, sql: str, *args): + self.calls.append(("fetchrow", sql[:60], args)) + if "kora_role_charter" in sql: + return self.role_charter_row + if "workspace_constitution_revisions" in sql: + return self.constitution_row + return None + + async def fetch(self, sql: str, *args): + self.calls.append(("fetch", sql[:60], args)) + if "ar.actor_kind = 'kora'" in sql: + return self.own_scratchpad_rows + if "ar.actor_kind != 'kora'" in sql: + return self.cross_agent_rows + if "LIKE 'kora.%'" in sql: + return self.recent_event_rows + return [] + + def transaction(self): + return _FakeTxn(self) + + +class _FakeAcquireCtx: + def __init__(self, conn): + self._conn = conn + + async def __aenter__(self): + return self._conn + + async def __aexit__(self, *exc): + return False + + +class _FakePool: + def __init__(self, conn): + self._conn = conn + + def acquire(self): + return _FakeAcquireCtx(self._conn) + + +WORKSPACE_ID = "org_test_workspace_001" + +_CONTENT_MD = "# Charter v1.0" +_CONTENT_HASH = hashlib.sha256(_CONTENT_MD.encode("utf-8")).hexdigest() +_JSONB = { + "schema_version": 1, + "charter_version": "1.0", + "sections": { + "identity": "Kora is bounded.", + "authority_can_do": ["a"], + "authority_cannot_do": ["b"], + "override_preconditions": ["c"], + "escalation_triggers": ["d"], + "per_session_discipline": ["e"], + "audit_attribution": "f", + "charter_modification": "g", + "effective_date_clause": "h", + }, +} + + +def _charter_row(): + return { + "id": "11111111-1111-1111-1111-111111111111", + "workspace_id": WORKSPACE_ID, + "schema_version": 1, + "content_md": _CONTENT_MD, + "content_jsonb": _JSONB, + "content_hash": _CONTENT_HASH, + "created_at": "2026-05-20T00:00:00Z", + } + + +# --------------------------------------------------------------------------- +# Assembler +# --------------------------------------------------------------------------- + + +def test_assemble_session_context_fans_out_six_reads(): + """All six reads fire (one fetchrow + four fetches + one fetchrow for constitution).""" + conn = _FakeConnection( + role_charter_row=_charter_row(), + own_scratchpad_rows=[], + cross_agent_rows=[], + recent_event_rows=[], + constitution_row={ + "revision_id": "22222222-2222-2222-2222-222222222222", + "rules_hash": b"\xab\xcd" * 16, + }, + ) + pool = _FakePool(conn) + ctx = asyncio.run(assemble_session_context(WORKSPACE_ID, pool)) + assert isinstance(ctx, KoraSessionContext) + assert ctx.workspace_id == WORKSPACE_ID + assert ctx.assembled_at # ISO-8601 non-empty + assert ctx.role_charter.charter_version == "1.0" + assert ctx.capability_matrix_row.actor_kind == "kora" + assert ctx.own_scratchpad == () + assert ctx.cross_agent_scratchpad == () + assert ctx.recent_chain_events == () + assert ctx.active_constitution_revision_id == "22222222-2222-2222-2222-222222222222" + assert ctx.active_constitution_rules_hash == "abcd" * 16 + + +def test_assemble_session_context_constitution_absent_returns_none_pair(): + """Fresh workspace (no Constitution revisions) → both fields are None.""" + conn = _FakeConnection(role_charter_row=_charter_row()) + pool = _FakePool(conn) + ctx = asyncio.run(assemble_session_context(WORKSPACE_ID, pool)) + assert ctx.active_constitution_revision_id is None + assert ctx.active_constitution_rules_hash is None + + +def test_assemble_session_context_default_limits(): + """Default scratchpad limit 100; default recent-event limit 50.""" + assert DEFAULT_SCRATCHPAD_LIMIT == 100 + assert DEFAULT_RECENT_EVENT_LIMIT == 50