diff --git a/.env.example b/.env.example index 03b0fb45..11af578d 100644 --- a/.env.example +++ b/.env.example @@ -15,6 +15,9 @@ POSTGRES_DB=lineageweave POSTGRES_PORT=15432 VALKEY_PORT=16379 +# Optional. 1 = fail-closed Valkey outbox (ADR 0026). Never invent a +# delivery. Default unset uses compose Valkey. +VALKEY_DISABLED= KEYCLOAK_ADMIN=admin KEYCLOAK_ADMIN_PASSWORD=admin_dev_only diff --git a/AGENTS.md b/AGENTS.md index c790995c..86797112 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -82,6 +82,11 @@ Gabriel factorization. Closest and farthest post–criterion pairs persist to `report_leftover_pair` and sit above the member list so a click opens that post. +Ticket activity is a transactional outbox (ADR 0026). Persist +`activity_outbox_event` first, then `XADD`. `GET /api/outbox` +fail-closes when Valkey is down. Never invent a stream id or a +theta. A hidden post is omitted from the home list. + `frontend/` has its own toolchain (Node pinned via `frontend/mise.toml`, pnpm via Corepack -- do not add a second Node package manager or a floating Node version): diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index f8a83ceb..c91be7af 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -398,7 +398,19 @@ create, `ticket_status_changed` on a status-changing `PATCH`) as the first real producer, and surfaced in the popup as an `ActivityPanel` (list + manual refresh) as the first real consumer. `make seed` XADDs `ticket_created` for the seeded A-100 and calendar tickets so Activity -is not empty after a report-member click. Verified against +is not empty after a report-member click. + +v0.85.0 (ADR 0026) adds the transactional outbox those dual-writes +were missing. `activity_outbox_event` is persisted as +`outbox_pending` before any `XADD`; a successful stream id flips the +row to `outbox_delivered`. `GET /api/outbox` fail-closes when +`VALKEY_DISABLED=1` or Valkey does not answer. Home Outbox sits +between Rankings and Calendar: unavailable copy is **Outbox · Valkey +not available**; an accepted delivery lists the summary and opens +that post. Hidden posts are omitted. Never invent a stream id or a +theta. TEPP's measurement outbox stays on #214. + +Verified against the actual Docker Compose network, not just `pytest`: created and patched a ticket through the real `backend` container talking to the real `valkey` container over the internal `redis://valkey:6379/0` DNS diff --git a/CHANGELOG.d/0.85.0-valkey-activity-outbox.md b/CHANGELOG.d/0.85.0-valkey-activity-outbox.md new file mode 100644 index 00000000..9c93e000 --- /dev/null +++ b/CHANGELOG.d/0.85.0-valkey-activity-outbox.md @@ -0,0 +1,9 @@ +# 0.85.0 — Fail-closed Valkey activity outbox + +## Added + +- Home Outbox panel lists durable ticket deliveries through + `ValkeyOutboxClient` (ADR 0026). After login with the port disabled + or Valkey down, Demo Analyst sees **Outbox · Valkey not available**. + An accepted delivery lists the summary; click opens that post. A + hidden post is omitted. Never invent a stream id or a theta. diff --git a/CHANGELOG.md b/CHANGELOG.md index 6bfcaa28..fbff401b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,16 @@ All notable changes to this project are documented here. Format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); versioning follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [0.85.0] - 2026-08-17 + +### Added + +- Home Outbox panel lists durable ticket deliveries through + `ValkeyOutboxClient` (ADR 0026). After login with the port disabled + or Valkey down, Demo Analyst sees **Outbox · Valkey not available**. + An accepted delivery lists the summary; click opens that post. A + hidden post is omitted. Never invent a stream id or a theta. + ## [0.75.0] - 2026-08-17 ### Added diff --git a/README.md b/README.md index a3626d9f..51b8652c 100644 --- a/README.md +++ b/README.md @@ -138,7 +138,8 @@ no ORM, no file DB) and to Keycloak's live JWKS for OIDC verification: make up make seed # scripts/seed_demo_data.py: inserts synthetic corp/account/post # rows keyed to the *real* Keycloak demo users' subject ids, - # plus Valkey ticket_created events so Activity is not empty + # plus Valkey ticket_created events and activity_outbox_event + # rows so Activity and home Outbox are not empty curl http://localhost:18420/healthz ``` diff --git a/backend/app/config.py b/backend/app/config.py index 68cad343..9efdc7c8 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -52,6 +52,10 @@ class Settings: # RankWeaveNotAvailable -- never invent a fused score. Default false # uses the in-process library already required by reconstruct.py. rankweave_disabled: bool + # Valkey activity outbox (ADR 0026). True = fail-closed + # ValkeyNotAvailable -- never invent a delivery. Default false + # uses the compose Valkey already required by activity_stream. + valkey_disabled: bool @property def keycloak_jwks_uri(self) -> str: @@ -88,4 +92,8 @@ def load_settings() -> Settings: .strip() .lower() in {"1", "true", "yes", "on"}, + valkey_disabled=os.environ.get("VALKEY_DISABLED", "") + .strip() + .lower() + in {"1", "true", "yes", "on"}, ) diff --git a/backend/app/main.py b/backend/app/main.py index 27f67911..5d7e9dfe 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -55,6 +55,7 @@ from lineageweave.post_summary import ContextualOrchestratorPostSummaryClient, NullPostSummaryClient from lineageweave.relation_verification import NullRelationVerificationClient, SearxngRelationVerificationClient from lineageweave.rankweave_client import build_rankweave_client +from lineageweave.valkey_outbox import ValkeyOutboxClient, build_valkey_outbox_client from backend.app.activity_stream import ( create_valkey_client, @@ -74,6 +75,11 @@ ) from backend.app.post_evaluation_ingestion import fetch_post_evaluation, ingest_post_evaluation from backend.app.ranking_ingestion import load_visible_ranking_posts +from backend.app.outbox_ingestion import ( + load_visible_outbox_rows, + mark_outbox_delivered, + persist_pending_outbox_event, +) from backend.app.report_ingestion import ( GROUPING_KINDS, fetch_period_comparison, @@ -242,6 +248,46 @@ def _rankweave_client(): return build_rankweave_client(disabled=load_settings().rankweave_disabled) +def _valkey_outbox_client(*, reachable: bool) -> ValkeyOutboxClient: + """Fail-closed unless Valkey answered this request (ADR 0026).""" + if load_settings().valkey_disabled or not reachable: + return build_valkey_outbox_client(disabled=True) + return ValkeyOutboxClient(ping=lambda: None) + + +async def _publish_outbox_activity( + pool: asyncpg.Pool, + valkey: redis.Redis, + post_id: str, + event_type_code: str, + actor_account_id: str, + event_summary: str, + issue_ticket_id: str | None = None, +) -> None: + """Persist pending, XADD, then mark delivered. Never invent an entry id.""" + async with pool.acquire() as conn: + outbox_event_id = await persist_pending_outbox_event( + conn, + post_id, + event_type_code, + actor_account_id, + event_summary, + issue_ticket_id=issue_ticket_id, + ) + if load_settings().valkey_disabled: + return + entry_id = await publish_activity_event( + valkey, + post_id, + event_type_code, + actor_account_id, + event_summary, + ) + async with pool.acquire() as conn: + await mark_outbox_delivered(conn, outbox_event_id, str(entry_id)) + + + def _can_see_post(account: CurrentAccount, post: asyncpg.Record) -> bool: """ABAC: public rows are visible; private rows require same-corp affiliation.""" if post["visibility_code"] == "public": @@ -971,12 +1017,14 @@ async def create_post_ticket( status.HTTP_422_UNPROCESSABLE_CONTENT, f"due_date {request.due_date!r} is not a valid YYYY-MM-DD date", ) from exc - await publish_activity_event( + await _publish_outbox_activity( + pool, valkey, post_id, "ticket_created", account.user_account_id, ticket_created_summary(request.ticket_title), + issue_ticket_id=str(ticket["issue_ticket_id"]), ) return ticket @@ -1029,7 +1077,8 @@ async def patch_ticket( if ticket is None: raise HTTPException(status.HTTP_404_NOT_FOUND, "ticket not found") if request.ticket_status_code is not None: - await publish_activity_event( + await _publish_outbox_activity( + pool, valkey, post_id, "ticket_status_changed", @@ -1037,6 +1086,7 @@ async def patch_ticket( ticket_status_changed_summary( ticket.get("ticket_status_label") or request.ticket_status_code ), + issue_ticket_id=issue_ticket_id, ) return ticket @@ -1102,12 +1152,14 @@ async def derive_post_commitment( status.HTTP_422_UNPROCESSABLE_CONTENT, f"due_date {commitment.due_date!r} is not a valid YYYY-MM-DD date", ) from exc - await publish_activity_event( + await _publish_outbox_activity( + pool, valkey, post_id, "commitment_derived", account.user_account_id, f"Commitment derived: {commitment.commitment_summary}", + issue_ticket_id=str(ticket["issue_ticket_id"]), ) return {"post_id": str(post["post_id"]), "has_commitment": True, "ticket": ticket} @@ -1150,3 +1202,31 @@ async def read_rankings( return _rankweave_client().as_api_payload( posts, can_see_post=lambda _row: True ) + + +@app.get("/api/outbox") +async def read_outbox( + account: CurrentAccount = Depends(get_current_account), + pool: asyncpg.Pool = Depends(get_pool), + valkey: redis.Redis = Depends(get_valkey), +) -> dict[str, Any]: + """Durable ticket deliveries (ADR 0026). + + Hidden posts are omitted. Never invents a stream id or a theta. + Fail-closed when Valkey is disabled or does not answer. + """ + _require_post_read(account) + reachable = False + if not load_settings().valkey_disabled: + try: + await valkey.ping() + reachable = True + except Exception: + reachable = False + async with pool.acquire() as conn: + rows = await load_visible_outbox_rows( + conn, lambda row: _can_see_post(account, row) + ) + return _valkey_outbox_client(reachable=reachable).as_api_payload( + rows, can_see_post=lambda _row: True + ) diff --git a/backend/app/outbox_ingestion.py b/backend/app/outbox_ingestion.py new file mode 100644 index 00000000..aecde14e --- /dev/null +++ b/backend/app/outbox_ingestion.py @@ -0,0 +1,162 @@ +"""Persist ticket activity on the transactional outbox, then publish. + +A hidden post is omitted from the buyer list. This module never invents +a Valkey entry id or a theta. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any, Callable, Mapping + +from lineageweave.valkey_outbox import DELIVERY_DELIVERED, DELIVERY_PENDING + +if TYPE_CHECKING: + import asyncpg + +__all__ = [ + "load_visible_outbox_rows", + "mark_outbox_delivered", + "persist_pending_outbox_event", + "persist_pending_outbox_event_sync", + "mark_outbox_delivered_sync", +] + + +_INSERT_PENDING = """ +insert into activity_outbox_event ( + post_id, issue_ticket_id, event_type_code, actor_account_id, + event_summary, delivery_status_code +) values ($1::uuid, $2::uuid, $3, $4::uuid, $5, $6) +on conflict (post_id, event_type_code, event_summary) do update + set issue_ticket_id = coalesce( + activity_outbox_event.issue_ticket_id, excluded.issue_ticket_id + ) +returning outbox_event_id, delivery_status_code, valkey_entry_id +""" + +_MARK_DELIVERED = """ +update activity_outbox_event + set delivery_status_code = $2, + valkey_entry_id = $3, + delivered_at = now() + where outbox_event_id = $1::uuid + and delivery_status_code = $4 + and valkey_entry_id is null +""" + +_LIST_ROWS = """ +select + outbox.outbox_event_id, + outbox.post_id, + post.post_title, + post.visibility_code, + post.corporate_entity_id, + outbox.event_type_code, + outbox.event_summary, + outbox.delivery_status_code, + outbox.valkey_entry_id, + outbox.requested_at + from activity_outbox_event as outbox + join source_post as post on post.post_id = outbox.post_id + order by outbox.requested_at desc, outbox.outbox_event_id +""" + + +async def persist_pending_outbox_event( + conn: "asyncpg.Connection", + post_id: str, + event_type_code: str, + actor_account_id: str, + event_summary: str, + issue_ticket_id: str | None = None, +) -> str: + """Insert a pending row. Idempotent on (post, type, summary).""" + row = await conn.fetchrow( + _INSERT_PENDING, + post_id, + issue_ticket_id, + event_type_code, + actor_account_id, + event_summary, + DELIVERY_PENDING, + ) + return str(row["outbox_event_id"]) + + +async def mark_outbox_delivered( + conn: "asyncpg.Connection", + outbox_event_id: str, + valkey_entry_id: str, +) -> None: + """Record the Valkey stream id. No-op when already delivered.""" + entry = str(valkey_entry_id or "").strip() + if not entry: + return + await conn.execute( + _MARK_DELIVERED, + outbox_event_id, + DELIVERY_DELIVERED, + entry, + DELIVERY_PENDING, + ) + + +async def load_visible_outbox_rows( + conn: "asyncpg.Connection", + can_see_post: Callable[[Mapping[str, Any]], bool], +) -> list[dict[str, Any]]: + """Read outbox rows the buyer may see. Hidden posts drop here.""" + rows = await conn.fetch(_LIST_ROWS) + return [dict(row) for row in rows if can_see_post(row)] + + +def persist_pending_outbox_event_sync( + cur: Any, + post_id: str, + event_type_code: str, + actor_account_id: str, + event_summary: str, + issue_ticket_id: str | None = None, +) -> str: + """psycopg2 twin of :func:`persist_pending_outbox_event` for ``make seed``.""" + cur.execute( + """ + insert into activity_outbox_event ( + post_id, issue_ticket_id, event_type_code, actor_account_id, + event_summary, delivery_status_code + ) values (%s, %s, %s, %s, %s, %s) + on conflict (post_id, event_type_code, event_summary) do update + set issue_ticket_id = coalesce( + activity_outbox_event.issue_ticket_id, excluded.issue_ticket_id + ) + returning outbox_event_id + """, + ( + post_id, + issue_ticket_id, + event_type_code, + actor_account_id, + event_summary, + DELIVERY_PENDING, + ), + ) + return str(cur.fetchone()[0]) + + +def mark_outbox_delivered_sync(cur: Any, outbox_event_id: str, valkey_entry_id: str) -> None: + """psycopg2 twin of :func:`mark_outbox_delivered` for ``make seed``.""" + entry = str(valkey_entry_id or "").strip() + if not entry: + return + cur.execute( + """ + update activity_outbox_event + set delivery_status_code = %s, + valkey_entry_id = %s, + delivered_at = now() + where outbox_event_id = %s + and delivery_status_code = %s + and valkey_entry_id is null + """, + (DELIVERY_DELIVERED, entry, outbox_event_id, DELIVERY_PENDING), + ) diff --git a/backend/tests/test_api.py b/backend/tests/test_api.py index 1db483ee..84d30412 100644 --- a/backend/tests/test_api.py +++ b/backend/tests/test_api.py @@ -30,6 +30,7 @@ _VALKEY_URL = os.environ.get("LINEAGEWEAVE_TEST_VALKEY_URL", "redis://localhost:16379/0") _REALM = "lineageweave-demo" _MIGRATION_PATH = Path(__file__).resolve().parents[2] / "migrations" / "0001_initial_schema.sql" +_OUTBOX_MIGRATION_PATH = Path(__file__).resolve().parents[2] / "migrations" / "0013_activity_outbox.sql" def _postgres_available() -> bool: @@ -112,6 +113,7 @@ def seeded_db(demo_analyst_token): try: with conn.cursor() as cur: cur.execute(_MIGRATION_PATH.read_text()) + cur.execute(_OUTBOX_MIGRATION_PATH.read_text()) cur.execute( "insert into common_lookup_value (lookup_category, lookup_code, lookup_label) values " "('corporate_entity_level', 'group', 'Group'), " diff --git a/backend/tests/test_config.py b/backend/tests/test_config.py index c2f3994d..c7dd9368 100644 --- a/backend/tests/test_config.py +++ b/backend/tests/test_config.py @@ -31,3 +31,13 @@ def test_rankweave_disabled_defaults_off(monkeypatch) -> None: def test_rankweave_disabled_flag_is_opt_in(monkeypatch) -> None: monkeypatch.setenv("RANKWEAVE_DISABLED", "1") assert load_settings().rankweave_disabled is True + + +def test_valkey_disabled_defaults_off(monkeypatch) -> None: + monkeypatch.delenv("VALKEY_DISABLED", raising=False) + assert load_settings().valkey_disabled is False + + +def test_valkey_disabled_flag_is_opt_in(monkeypatch) -> None: + monkeypatch.setenv("VALKEY_DISABLED", "1") + assert load_settings().valkey_disabled is True diff --git a/docker/postgres-init/Dockerfile b/docker/postgres-init/Dockerfile index 51ac998c..7665dcbe 100644 --- a/docker/postgres-init/Dockerfile +++ b/docker/postgres-init/Dockerfile @@ -19,6 +19,7 @@ COPY migrations/0009_shared_metric_bank.sql /docker-entrypoint-initdb.d/10-share COPY migrations/0010_report_item_information.sql /docker-entrypoint-initdb.d/11-report-item-information.sql COPY migrations/0011_post_chat_result.sql /docker-entrypoint-initdb.d/12-post-chat-result.sql COPY migrations/0012_report_leftover_pair.sql /docker-entrypoint-initdb.d/13-report-leftover-pair.sql +COPY migrations/0013_activity_outbox.sql /docker-entrypoint-initdb.d/14-activity-outbox.sql # Official image already drops to this account at runtime; declare it so # the Dockerfile itself satisfies DS-0002 (explicit non-root USER). USER postgres diff --git a/docs/adr/0026-valkey-activity-outbox.md b/docs/adr/0026-valkey-activity-outbox.md new file mode 100644 index 00000000..d2eb791a --- /dev/null +++ b/docs/adr/0026-valkey-activity-outbox.md @@ -0,0 +1,49 @@ +# ADR 0026 — Fail-closed Valkey transactional activity outbox + +**Decision status:** Accepted +**Date:** 2026-08-17 + +## Context + +Ticket create/status-change already `XADD`s onto `activity:{post_id}` +(Phase 5b). That is a dual-write: Postgres commits the ticket, then +Valkey receives the event. If the stream write fails after the +ticket row exists, Activity is empty and there is no durable retry +evidence (Hohpe & Woolf, 2003, Transactional Outbox; Kleppmann, +2017). Milestone 2 (#79 / #87) requires outbox delivery and +idempotency evidence without a second application. + +TEPP's own outbox lives on #214 (ADR 0023). This ADR is the +activity-queue outbox on protected `main`. It does not invent a +fused score or a theta. + +## Decision + +1. Persist `activity_outbox_event` (3NF, two-or-more-word + `snake_case`) as `outbox_pending` before any `XADD`. +2. After a successful stream write, store the Valkey entry id and + flip the row to `outbox_delivered`. A missing stream id is not a + delivery. +3. `GET /api/outbox` (`post_read`) fail-closes when + `VALKEY_DISABLED=1` or Valkey does not answer. Unavailable copy + is **Outbox · Valkey not available**. Accepted rows list the + event summary; click opens that `source_post`. Hidden posts are + omitted. +4. Idempotency is `(post_id, event_type_code, event_summary)`. + Re-seed does not invent a second delivery. + +## Consequences + +`make seed` writes the pending row, `XADD`s, then marks delivered +so home Outbox is not empty after a fresh stack. Activity still +reads the stream. RankWeave stays on ADR 0024. Leftover pairs stay +on ADR 0017 / 0018. TEPP stays on #214. + +## References + +Hohpe, G., & Woolf, B. (2003). *Enterprise integration patterns: +Designing, building, and deploying messaging solutions*. +Addison-Wesley. + +Kleppmann, M. (2017). *Designing data-intensive applications*. +O'Reilly Media. diff --git a/docs/lineage-bi-research-notes.md b/docs/lineage-bi-research-notes.md index acd55620..a0b85329 100644 --- a/docs/lineage-bi-research-notes.md +++ b/docs/lineage-bi-research-notes.md @@ -340,3 +340,21 @@ model answers using only those sources and reports which ones it actually drew from). This is the Agentic retrieve-reason-cite shape the product brief asks for without adding a full agent-framework dependency for what two functions and a structured prompt already do. + +## Transactional activity outbox (ADR 0026) + +Ticket create and status-change used to be a dual-write: Postgres +committed the `issue_ticket` row, then Valkey received an `XADD`. +Hohpe and Woolf (2003) document the Transactional Outbox so a +message is not lost when the second write fails; Kleppmann (2017) +describes the same dual-write hazard. `activity_outbox_event` is +the durable row. A missing Valkey port is `valkey_not_available`, +never an invented stream id or a theta. Hidden posts stay out of +`GET /api/outbox`. TEPP measurement stays on TEPP's own contract. + +Hohpe, G., & Woolf, B. (2003). *Enterprise integration patterns: +Designing, building, and deploying messaging solutions*. +Addison-Wesley. + +Kleppmann, M. (2017). *Designing data-intensive applications*. +O'Reilly Media. diff --git a/frontend/package.json b/frontend/package.json index 575b7c58..c8f67bc8 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -1,7 +1,7 @@ { "name": "frontend", "private": true, - "version": "0.75.0", + "version": "0.85.0", "type": "module", "scripts": { "dev": "vite", diff --git a/frontend/src/App.test.tsx b/frontend/src/App.test.tsx index a32a2640..816bbe1e 100644 --- a/frontend/src/App.test.tsx +++ b/frontend/src/App.test.tsx @@ -65,6 +65,17 @@ describe("App, authenticated", () => { fused_rank: number; }[]; }; + outbox?: { + status?: "accepted" | "unavailable"; + status_reason?: string | null; + deliveries?: { + post_id: string; + post_title: string; + event_summary: string; + delivery_status_code: string; + valkey_entry_id: string; + }[]; + }; chatUnavailable?: boolean; searchUnavailable?: boolean; verificationEvidenceUrl?: string | null; @@ -228,6 +239,21 @@ describe("App, authenticated", () => { }), ); } + if (url.endsWith("/api/outbox")) { + const outbox = options?.outbox ?? { + status: "unavailable" as const, + status_reason: "valkey_not_available", + deliveries: [], + }; + return Promise.resolve( + jsonResponse({ + port: "valkey", + status: outbox.status, + status_reason: outbox.status_reason, + deliveries: outbox.deliveries ?? [], + }), + ); + } if (url.includes("/api/reports/compare/") && method === "GET") { return Promise.resolve( jsonResponse({ @@ -1307,6 +1333,49 @@ describe("App, authenticated", () => { await waitFor(() => expect(screen.getByText("The full body text.")).toBeInTheDocument()); }); + it("names Valkey unavailability on home outbox instead of inventing a delivery", async () => { + stubBackend(); + render(); + + expect(await screen.findByText("Outbox · Valkey not available")).toBeInTheDocument(); + expect( + screen.queryByText("Ticket created: Send Northridge Grid the revised quote"), + ).not.toBeInTheDocument(); + }); + + it("opens an accepted outbox delivery without inventing a stream id", async () => { + stubBackend({ + outbox: { + status: "accepted", + status_reason: null, + deliveries: [ + { + post_id: "post-1", + post_title: "Public post", + event_summary: "Ticket created: Send Northridge Grid the revised quote", + delivery_status_code: "outbox_delivered", + valkey_entry_id: "1-0", + }, + ], + }, + }); + render(); + + const outboxButton = await screen.findByRole("button", { + name: /open outbox: ticket created: send northridge grid the revised quote/i, + }); + expect(outboxButton).toHaveTextContent("Ticket created: Send Northridge Grid the revised quote"); + expect(outboxButton).toHaveTextContent("Outbox · valkey"); + expect(outboxButton).toHaveTextContent("Public post"); + expect( + screen.queryByRole("button", { name: /open outbox: ticket created: hidden/i }), + ).not.toBeInTheDocument(); + + await userEvent.click(outboxButton); + + await waitFor(() => expect(screen.getByText("The full body text.")).toBeInTheDocument()); + }); + it("shows upcoming commitments on the home page calendar and opens the post on click", async () => { stubBackend(); render(); diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index 6056e5eb..9fe9c277 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -26,6 +26,7 @@ import { fetchPeriodReports, fetchPosts, fetchRankings, + fetchOutbox, fetchRelatedEntity, fetchRelatedKeymen, rebuildLineage, @@ -51,6 +52,7 @@ import { type PostLineage, type PostSummary, type RankingList, + type OutboxList, type RelatedNode, type VocEvidence, } from "./api"; @@ -1371,6 +1373,64 @@ function RankingsPanel({ ); } +function OutboxPanel({ + accessToken, + onSelectPost, +}: { + accessToken: string; + onSelectPost: (postId: string) => void; +}) { + const [outbox, setOutbox] = useState(null); + const [error, setError] = useState(null); + + useEffect(() => { + setError(null); + fetchOutbox(accessToken) + .then(setOutbox) + .catch((err) => setError(String(err))); + }, [accessToken]); + + return ( +
+
+

Outbox

+ {outbox && ( + + {outbox.status === "accepted" + ? "valkey" + : `valkey · ${outbox.status_reason ?? "unavailable"}`} + + )} +
+ {error &&

{error}

} + {outbox === null && !error &&

Loading outbox...

} + {outbox && outbox.status === "unavailable" && ( +

Outbox · Valkey not available

+ )} + {outbox && outbox.status === "accepted" && outbox.deliveries.length === 0 && ( +

No delivered ticket events on Valkey.

+ )} + {outbox && outbox.deliveries.length > 0 && ( +
    + {outbox.deliveries.map((hit) => ( +
  • + +
    • + ))} +
+ )} +
+ ); +} + function CalendarPanel({ accessToken, onSelectPost, @@ -1690,6 +1750,7 @@ function PostList({ accessToken }: { accessToken: string }) { return ( <> +
diff --git a/frontend/src/api.ts b/frontend/src/api.ts index e6dfcbad..207a7425 100644 --- a/frontend/src/api.ts +++ b/frontend/src/api.ts @@ -515,3 +515,22 @@ export interface RankingList { export function fetchRankings(accessToken: string): Promise { return backendFetch("/api/rankings", accessToken); } + +export interface OutboxDelivery { + post_id: string; + post_title: string; + event_summary: string; + delivery_status_code: string; + valkey_entry_id: string; +} + +export interface OutboxList { + port: string; + status: "accepted" | "unavailable"; + status_reason: string | null; + deliveries: OutboxDelivery[]; +} + +export function fetchOutbox(accessToken: string): Promise { + return backendFetch("/api/outbox", accessToken); +} diff --git a/lineageweave/__init__.py b/lineageweave/__init__.py index 1710c009..019d40ad 100644 --- a/lineageweave/__init__.py +++ b/lineageweave/__init__.py @@ -35,4 +35,4 @@ "sentence_excerpts", ] -__version__ = "0.75.0" +__version__ = "0.85.0" diff --git a/lineageweave/valkey_outbox.py b/lineageweave/valkey_outbox.py new file mode 100644 index 00000000..87b561c9 --- /dev/null +++ b/lineageweave/valkey_outbox.py @@ -0,0 +1,146 @@ +"""Fail-closed adapter for the Valkey transactional activity outbox. + +Ticket mutations persist an ``activity_outbox_event`` row first, then +``XADD`` onto ``activity:{post_id}``. A missing or disabled Valkey +port must not invent a stream id, a delivery, or a theta (Hohpe & +Woolf, 2003; Kleppmann, 2017). Hidden posts are omitted from the +buyer projection. + +This module does not replace ``backend.app.activity_stream`` and does +not implement TEPP measurement (ADR 0022 on #214). +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Callable, Mapping, Sequence + +DELIVERY_PENDING = "outbox_pending" +DELIVERY_DELIVERED = "outbox_delivered" +DELIVERY_FAILED = "outbox_failed" +PORT_NAME = "valkey" + + +class ValkeyNotAvailable(RuntimeError): + """Raised when the Valkey outbox port is down or disabled.""" + + reason = "valkey_not_available" + + +def _no_ping() -> None: + raise ValkeyNotAvailable( + "valkey_not_available: Valkey outbox port is not configured. " + "Pass VALKEY_DISABLED=0 (default) or a ping= callable. " + "Never invent a delivery." + ) + + +@dataclass(frozen=True) +class OutboxDelivery: + """One visible delivered event. Never a calibrated theta.""" + + post_id: str + post_title: str + event_summary: str + delivery_status_code: str + valkey_entry_id: str + + def to_json(self) -> dict[str, Any]: + return { + "post_id": self.post_id, + "post_title": self.post_title, + "event_summary": self.event_summary, + "delivery_status_code": self.delivery_status_code, + "valkey_entry_id": self.valkey_entry_id, + } + + +@dataclass(frozen=True) +class OutboxList: + """Accepted outbox projection. Empty when nothing was delivered.""" + + items: tuple[OutboxDelivery, ...] + + def to_json(self) -> list[dict[str, Any]]: + return [item.to_json() for item in self.items] + + +def project_outbox_list( + rows: Sequence[Mapping[str, Any]], + can_see_post: Callable[[Mapping[str, Any]], bool], +) -> OutboxList: + """Accept persisted rows. Pending/failed/hidden rows drop. No invented id.""" + items: list[OutboxDelivery] = [] + seen: set[str] = set() + for row in rows: + if not can_see_post(row): + continue + status = str(row.get("delivery_status_code") or "").strip() + post_id = str(row.get("post_id") or "").strip() + title = str(row.get("post_title") or "").strip() + summary = str(row.get("event_summary") or "").strip() + entry_id = str(row.get("valkey_entry_id") or "").strip() + if status != DELIVERY_DELIVERED: + continue + if not post_id or not title or not summary or not entry_id: + continue + dedupe = f"{post_id}:{summary}" + if dedupe in seen: + continue + seen.add(dedupe) + items.append( + OutboxDelivery( + post_id=post_id, + post_title=title, + event_summary=summary, + delivery_status_code=status, + valkey_entry_id=entry_id, + ) + ) + return OutboxList(items=tuple(items)) + + +def build_valkey_outbox_client( + disabled: bool = False, + ping: Callable[[], None] | None = None, +) -> "ValkeyOutboxClient": + """``disabled=True`` keeps the default fail-closed ping.""" + if disabled: + return ValkeyOutboxClient() + if ping is None: + return ValkeyOutboxClient() + return ValkeyOutboxClient(ping=ping) + + +class ValkeyOutboxClient: + """Projects durable outbox rows only when Valkey itself answers.""" + + def __init__(self, ping: Callable[[], None] = _no_ping) -> None: + self._ping = ping + + def as_api_payload( + self, + rows: Sequence[Mapping[str, Any]], + can_see_post: Callable[[Mapping[str, Any]], bool], + ) -> dict[str, Any]: + """Buyer-visible outbox status. Never invents a delivery.""" + try: + self._ping() + except ValkeyNotAvailable: + return { + "port": PORT_NAME, + "status": "unavailable", + "status_reason": ValkeyNotAvailable.reason, + "deliveries": [], + } + except Exception as exc: + raise ValkeyNotAvailable( + f"valkey_not_available: outbox ping failed ({exc})" + ) from exc + listing = project_outbox_list(rows, can_see_post) + return { + "port": PORT_NAME, + "status": "accepted", + "status_reason": None, + "deliveries": listing.to_json(), + } diff --git a/migrations/0013_activity_outbox.sql b/migrations/0013_activity_outbox.sql new file mode 100644 index 00000000..5194485e --- /dev/null +++ b/migrations/0013_activity_outbox.sql @@ -0,0 +1,37 @@ +-- ADR 0026: transactional activity outbox. Persist the ticket event in +-- PostgreSQL first, then XADD onto Valkey. CREATE IF NOT EXISTS so a +-- volume that already ran 0001 still upgrades. Lookup codes are unique +-- across categories (see 0001). Never store a fused score or a theta. + +insert into common_lookup_value (lookup_category, lookup_code, lookup_label, display_order) +values + ('activity_event_type', 'ticket_created', 'Ticket created', 0), + ('activity_event_type', 'ticket_status_changed', 'Ticket status changed', 1), + ('activity_event_type', 'commitment_derived', 'Commitment derived', 2), + ('outbox_delivery_status', 'outbox_pending', 'Pending', 0), + ('outbox_delivery_status', 'outbox_delivered', 'Delivered', 1), + ('outbox_delivery_status', 'outbox_failed', 'Failed', 2) +on conflict (lookup_code) do nothing; + +create table if not exists activity_outbox_event ( + outbox_event_id uuid primary key default uuid_generate_v4(), + post_id uuid not null references source_post (post_id), + issue_ticket_id uuid references issue_ticket (issue_ticket_id), + event_type_code text not null references common_lookup_value (lookup_code), + actor_account_id uuid not null references user_account (user_account_id), + event_summary text not null, + delivery_status_code text not null references common_lookup_value (lookup_code), + valkey_entry_id text, + requested_at timestamptz not null default now(), + delivered_at timestamptz, + unique (post_id, event_type_code, event_summary), + check ( + (delivery_status_code = 'outbox_delivered' and valkey_entry_id is not null and delivered_at is not null) + or (delivery_status_code <> 'outbox_delivered' and valkey_entry_id is null) + ) +); + +create index if not exists activity_outbox_event_post_idx + on activity_outbox_event (post_id); +create index if not exists activity_outbox_event_status_idx + on activity_outbox_event (delivery_status_code, requested_at desc); diff --git a/pyproject.toml b/pyproject.toml index 764ebad7..ea3bccd8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "lineageweave" -version = "0.75.0" +version = "0.85.0" description = "Reconstructs git-branch-style lineage DAGs from scattered short records using multi-channel score fusion and LLM adjudication." readme = "README.md" license = { text = "MIT" } diff --git a/scripts/seed_demo_data.py b/scripts/seed_demo_data.py index d4c24d84..94c2b823 100644 --- a/scripts/seed_demo_data.py +++ b/scripts/seed_demo_data.py @@ -107,6 +107,7 @@ def seed( cur.execute((migrations / "0010_report_item_information.sql").read_text()) cur.execute((migrations / "0011_post_chat_result.sql").read_text()) cur.execute((migrations / "0012_report_leftover_pair.sql").read_text()) + cur.execute((migrations / "0013_activity_outbox.sql").read_text()) cur.execute( """ insert into common_lookup_value (lookup_category, lookup_code, lookup_label, display_order) values @@ -789,11 +790,11 @@ def _seed_fixture_tickets(cur) -> None: def _seed_fixture_ticket_activity(cur, actor_account_id, valkey_url: str) -> None: - """``XADD`` ticket_created onto each seeded ticket's post stream. + """Persist pending outbox rows, then ``XADD`` ticket_created. - Without this, GET /api/posts/{id}/activity is empty after ``make seed`` - even though the ticket row exists -- Activity reads Valkey, not - Postgres. Idempotent: a matching summary on that stream is left alone. + Without the outbox row, GET /api/outbox is empty after ``make seed`` + even though Activity can read Valkey. Without the stream write, + GET /api/posts/{id}/activity is empty. Idempotent on summary. """ try: import redis @@ -806,6 +807,10 @@ def _seed_fixture_ticket_activity(cur, actor_account_id, valkey_url: str) -> Non publish_activity_event_sync, ticket_created_summary, ) + from backend.app.outbox_ingestion import ( + mark_outbox_delivered_sync, + persist_pending_outbox_event_sync, + ) from lineageweave.fixtures import ambiguous_commitment_post specs = [(title, ticket) for title, ticket, _due in FIXTURE_TICKET_SPECS] @@ -821,18 +826,31 @@ def _seed_fixture_ticket_activity(cur, actor_account_id, valkey_url: str) -> Non if row is None: continue cur.execute( - "select 1 from issue_ticket where post_id = %s and ticket_title = %s", + "select issue_ticket_id from issue_ticket " + "where post_id = %s and ticket_title = %s", (row[0], ticket_title), ) - if cur.fetchone() is None: + ticket_row = cur.fetchone() + if ticket_row is None: continue - publish_activity_event_sync( + summary = ticket_created_summary(ticket_title) + outbox_event_id = persist_pending_outbox_event_sync( + cur, + str(row[0]), + "ticket_created", + str(actor_account_id), + summary, + issue_ticket_id=str(ticket_row[0]), + ) + entry_id = publish_activity_event_sync( client, str(row[0]), "ticket_created", str(actor_account_id), - ticket_created_summary(ticket_title), + summary, ) + if entry_id: + mark_outbox_delivered_sync(cur, outbox_event_id, str(entry_id)) except redis.RedisError as exc: raise SystemExit( f"Valkey at {valkey_url} is unreachable -- did you run `make up`? ({exc})" diff --git a/tests/test_activity_outbox_schema.py b/tests/test_activity_outbox_schema.py new file mode 100644 index 00000000..38f9aeb4 --- /dev/null +++ b/tests/test_activity_outbox_schema.py @@ -0,0 +1,99 @@ +"""Real-database contract for migrations/0013_activity_outbox.sql. + +Applies 0001 then 0013 on a throwaway database. Self-skips without +PostgreSQL. Proves the outbox table, lookup codes, and the delivered +row's required stream id -- never a fabricated theta. +""" + +from __future__ import annotations + +import os +import uuid +from pathlib import Path + +import psycopg2 +import pytest + +_ADMIN_DSN = os.environ.get( + "LINEAGEWEAVE_TEST_POSTGRES_ADMIN_DSN", "postgresql://localhost/postgres" +) +_ROOT = Path(__file__).resolve().parents[1] +_MIGRATION_0001 = _ROOT / "migrations" / "0001_initial_schema.sql" +_MIGRATION_0013 = _ROOT / "migrations" / "0013_activity_outbox.sql" + + +def _postgres_available() -> bool: + try: + conn = psycopg2.connect(_ADMIN_DSN, connect_timeout=2) + conn.close() + return True + except psycopg2.OperationalError: + return False + + +pytestmark = pytest.mark.skipif( + not _postgres_available(), + reason=f"no reachable PostgreSQL server at {_ADMIN_DSN}", +) + + +@pytest.fixture +def outbox_schema_db(): + db_name = f"lineageweave_outbox_{uuid.uuid4().hex[:12]}" + admin_conn = psycopg2.connect(_ADMIN_DSN) + admin_conn.autocommit = True + with admin_conn.cursor() as cur: + cur.execute(f'create database "{db_name}"') + try: + db_dsn = _ADMIN_DSN.rsplit("/", 1)[0] + f"/{db_name}" + conn = psycopg2.connect(db_dsn) + try: + with conn.cursor() as cur: + cur.execute(_MIGRATION_0001.read_text()) + cur.execute(_MIGRATION_0013.read_text()) + conn.commit() + yield conn + finally: + conn.close() + finally: + with admin_conn.cursor() as cur: + cur.execute(f'drop database "{db_name}"') + admin_conn.close() + + +def test_activity_outbox_table_and_lookups_exist(outbox_schema_db) -> None: + with outbox_schema_db.cursor() as cur: + cur.execute( + "select 1 from information_schema.tables " + "where table_schema = 'public' and table_name = 'activity_outbox_event'" + ) + assert cur.fetchone() is not None + cur.execute( + "select lookup_code from common_lookup_value " + "where lookup_category in ('activity_event_type', 'outbox_delivery_status') " + "order by lookup_code" + ) + codes = {row[0] for row in cur.fetchall()} + assert { + "ticket_created", + "ticket_status_changed", + "commitment_derived", + "outbox_pending", + "outbox_delivered", + "outbox_failed", + } <= codes + + +def test_delivered_row_requires_a_stream_id(outbox_schema_db) -> None: + with outbox_schema_db.cursor() as cur: + cur.execute( + """ + select pg_get_constraintdef(oid) + from pg_constraint + where conrelid = 'activity_outbox_event'::regclass + and contype = 'c' + """ + ) + checks = " ".join(row[0] for row in cur.fetchall()) + assert "outbox_delivered" in checks + assert "valkey_entry_id" in checks diff --git a/tests/test_valkey_outbox.py b/tests/test_valkey_outbox.py new file mode 100644 index 00000000..fb5f039d --- /dev/null +++ b/tests/test_valkey_outbox.py @@ -0,0 +1,98 @@ +"""Fail-closed Valkey transactional outbox. + +A missing or disabled Valkey port never invents a stream id, a +delivery, or a theta. Hidden posts drop from the buyer projection. +""" + +from __future__ import annotations + +import pytest + +from lineageweave.valkey_outbox import ( + ValkeyNotAvailable, + ValkeyOutboxClient, + build_valkey_outbox_client, + project_outbox_list, +) + +PUBLIC = { + "post_id": "post-1", + "post_title": "Public post", + "visibility_code": "public", + "event_summary": "Ticket created: Send Northridge Grid the revised quote", + "delivery_status_code": "outbox_delivered", + "valkey_entry_id": "1-0", +} +HIDDEN = { + "post_id": "hidden-parent", + "post_title": "Private parent", + "visibility_code": "private", + "event_summary": "Ticket created: hidden ticket", + "delivery_status_code": "outbox_delivered", + "valkey_entry_id": "1-1", +} +PENDING = { + "post_id": "post-1", + "post_title": "Public post", + "visibility_code": "public", + "event_summary": "Ticket created: pending only", + "delivery_status_code": "outbox_pending", + "valkey_entry_id": None, +} + + +def test_default_payload_never_invents_a_delivery() -> None: + payload = ValkeyOutboxClient().as_api_payload( + [PUBLIC], + can_see_post=lambda _row: True, + ) + assert payload == { + "port": "valkey", + "status": "unavailable", + "status_reason": "valkey_not_available", + "deliveries": [], + } + + +def test_disabled_factory_fails_closed() -> None: + client = build_valkey_outbox_client(disabled=True, ping=lambda: None) + payload = client.as_api_payload([PUBLIC], can_see_post=lambda _row: True) + assert payload["status"] == "unavailable" + assert payload["deliveries"] == [] + + +def test_reachable_port_lists_visible_delivered_rows() -> None: + payload = ValkeyOutboxClient(ping=lambda: None).as_api_payload( + [PUBLIC, HIDDEN, PENDING], + can_see_post=lambda row: row["post_id"] != "hidden-parent", + ) + assert payload["status"] == "accepted" + assert payload["status_reason"] is None + assert payload["deliveries"] == [ + { + "post_id": "post-1", + "post_title": "Public post", + "event_summary": "Ticket created: Send Northridge Grid the revised quote", + "delivery_status_code": "outbox_delivered", + "valkey_entry_id": "1-0", + } + ] + + +def test_project_drops_pending_and_missing_stream_id() -> None: + listing = project_outbox_list( + [PENDING, {**PUBLIC, "valkey_entry_id": ""}], + can_see_post=lambda _row: True, + ) + assert listing.to_json() == [] + + +def test_ping_exception_fails_closed() -> None: + def boom() -> None: + raise RuntimeError("connection refused") + + with pytest.raises(ValkeyNotAvailable, match="valkey_not_available"): + ValkeyOutboxClient(ping=boom).as_api_payload( + [PUBLIC], + can_see_post=lambda _row: True, + ) diff --git a/uv.lock b/uv.lock index 08eab776..a5e94ad6 100644 --- a/uv.lock +++ b/uv.lock @@ -355,7 +355,7 @@ wheels = [ [[package]] name = "lineageweave" -version = "0.75.0" +version = "0.85.0" source = { virtual = "." } dependencies = [ { name = "certifi" },