diff --git a/AGENTS.md b/AGENTS.md index d43bdd733..e81f9965a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -44,11 +44,14 @@ does it (`gh repo list ContextualWisdomLab`). ## Pluggable channels: never fake a missing signal -`NullEmbeddingClient` and `NullAdjudicationClient` (and any new channel -client you add) must set `available = False` and make their channel dropped -+ renormalized (`reconstruct.active_weights`), never silently return a -placeholder score. A missing signal and a confidently-negative signal are -different things. +`NullEmbeddingClient`, `NullAdjudicationClient`, and +`NullKeymanExtractionClient` (and any new channel client you add) must +set `available = False` and make their channel dropped + renormalized +(`reconstruct.active_weights`), never silently return a placeholder +score or invented Keyman. A missing signal and a confidently-negative +signal are different things. Keyman extraction goes through +contextual-orchestrator the same way adjudication does -- never a raw +LLM API. ## Tests diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index b19a903f0..8eb91782f 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -168,6 +168,17 @@ stack itself with the same shape of synthetic data for manual/frontend use. `CORSMiddleware` (`backend/app/main.py`) allows exactly the frontend's origin(s) (`FRONTEND_ORIGINS`), `GET` only, `Authorization` header only. +Phase 2 adds two more GET endpoints on the same RBAC+ABAC gate plus one +write: `GET /api/posts/{post_id}/keymen` (people extracted or seeded for +that post, with N:N affiliations), `GET /api/keymen/{person_id}/related` +(RWR from that person over `knowledge_graph_edge` rows, adaptive +relevance cutoff, never a fixed hop count), and +`POST /api/posts/{post_id}/extract-keymen` (`post_admin` only -- a write +with a real LLM-call cost). A Keyman who is only mentioned on a post the +account cannot see is 403, matching the post deny path. Extraction +lives in `lineageweave/keyman_extraction.py` and talks to +contextual-orchestrator; persist is `backend/app/keyman_ingestion.py`. + ### Frontend (`frontend/`) React + Vite + TypeScript, pinned Node via `mise.toml`, pnpm via Corepack. diff --git a/CHANGELOG.md b/CHANGELOG.md index 778b33859..00734eec6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,32 @@ 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.9.0] - 2026-08-13 + +### Added + +- Milestone 4, Phase 2 continues: `lineageweave/keyman_extraction.py` + extracts two-sided Keymen (our-side vs counterparty, 0..N organization + affiliations) from a post's title+body. The live client calls + contextual-orchestrator with `mode="route"` -- never a raw LLM API. + `NullKeymanExtractionClient` stays unavailable rather than inventing + mentions. +- `lineageweave.knowledge_graph.knowledge_graph_edges_for_post` populates + the three Phase 2 edge kinds (person<->post mention, person<->corporate + entity affiliation, person<->person co-mention) as typed + `knowledge_graph_edge` specs. RWR then runs on that graph. +- Backend endpoints `GET /api/posts/{post_id}/keymen`, + `GET /api/keymen/{person_id}/related`, and + `POST /api/posts/{post_id}/extract-keymen`, RBAC+ABAC-gated the same + way as post detail: a Keyman who is only mentioned on another corp's + private post 403s, related-node traversal never returns those hidden + posts, and extraction is `post_admin` (a write with a real LLM cost). + Persist goes through `backend/app/keyman_ingestion.py` into + `person` / `person_affiliation` / `post_person_mention` / + `knowledge_graph_edge`. +- `fixtures.ambiguous_keyman_post`: a synthetic, non-templated workshop + follow-up used by the parser tests and the real-orchestrator Keyman test. + ## [0.8.0] - 2026-08-13 ### Added diff --git a/README.md b/README.md index 6371689a3..ee06413d2 100644 --- a/README.md +++ b/README.md @@ -141,13 +141,17 @@ make seed # scripts/seed_demo_data.py: inserts synthetic corp/account/post curl http://localhost:18420/healthz ``` -`GET /api/posts` and `GET /api/posts/{post_id}` require a real bearer token -(RBAC: the account's role must grant `post_read`; ABAC: a private post is -only visible to accounts affiliated with its owning corporate entity -- -`backend/app/main.py`). `backend/tests/test_api.py` proves both the allow -and the deny path against a live Keycloak + throwaway Postgres database, -including that a private post scoped to a *different* corporate entity is -excluded from the list and 403s on direct fetch. +`GET /api/posts`, `GET /api/posts/{post_id}`, +`GET /api/posts/{post_id}/keymen`, `GET /api/keymen/{person_id}/related`, +and `POST /api/posts/{post_id}/extract-keymen` +require a real bearer token (RBAC: the account's role must grant +`post_read`; ABAC: a private post is only visible to accounts affiliated +with its owning corporate entity -- `backend/app/main.py`). A Keyman who +is only mentioned on a post the account cannot see is 403, same deny +path. `backend/tests/test_api.py` proves both the allow and the deny +path against a live Keycloak + throwaway Postgres database, including +that a private post scoped to a *different* corporate entity is excluded +from the list and 403s on direct fetch. `frontend/` (React + Vite + TypeScript, `docker compose`'s fourth service) is a real client, not mocked or static: `react-oidc-context` drives an diff --git a/backend/app/config.py b/backend/app/config.py index 4bb88ad17..0b41ab760 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -26,6 +26,11 @@ class Settings: # host-published port a browser actually hits). keycloak_issuer: str frontend_origins: list[str] + # Keyman extraction is a hard dependency of POST /api/posts/{id}/extract-keymen + # only -- every other endpoint works with these unset. Empty string, not + # a fabricated default, when unconfigured (see keyman_ingestion.py). + orchestrator_base_url: str + orchestrator_api_key: str @property def keycloak_jwks_uri(self) -> str: @@ -51,4 +56,6 @@ def load_settings() -> Settings: for origin in os.environ.get("FRONTEND_ORIGINS", "http://localhost:5173").split(",") if origin.strip() ], + orchestrator_base_url=os.environ.get("ORCHESTRATOR_BASE_URL", ""), + orchestrator_api_key=os.environ.get("ORCHESTRATOR_API_KEY", ""), ) diff --git a/backend/app/keyman_ingestion.py b/backend/app/keyman_ingestion.py new file mode 100644 index 000000000..edecd19af --- /dev/null +++ b/backend/app/keyman_ingestion.py @@ -0,0 +1,84 @@ +"""Runs a `KeymanExtractionClient` over a post and persists the result: +`person` (upserted by name + side -- the schema has no unique constraint +on person_name alone, since two different real people can share a name, +but re-running extraction on the same post should not keep creating +duplicate rows for the same extracted mention), `person_affiliation` +(N:N, matched to a real `corporate_entity` by exact case-insensitive name +where possible), and `post_person_mention`. Finishes by calling +`knowledge_graph.persist_edges_for_post` so the Knowledge Graph edges are +computed from the same write, not a separate manual step. +""" + +from __future__ import annotations + +import asyncpg + +from lineageweave.keyman_extraction import KeymanExtractionClient, PersonMention + +from .knowledge_graph import persist_edges_for_post + + +async def _resolve_corporate_entity_id(conn: asyncpg.Connection, organization_name: str) -> str | None: + row = await conn.fetchrow( + "select corporate_entity_id from corporate_entity where lower(entity_name) = lower($1)", + organization_name, + ) + return str(row["corporate_entity_id"]) if row is not None else None + + +async def _upsert_person(conn: asyncpg.Connection, mention: PersonMention) -> str: + row = await conn.fetchrow( + "select person_id from person where person_name = $1 and person_side_code = $2", + mention.person_name, + mention.person_side_code, + ) + if row is not None: + return str(row["person_id"]) + row = await conn.fetchrow( + "insert into person (person_name, person_side_code) values ($1, $2) returning person_id", + mention.person_name, + mention.person_side_code, + ) + return str(row["person_id"]) + + +async def ingest_post_keymen( + conn: asyncpg.Connection, + client: KeymanExtractionClient, + post_id: str, + post_title: str, + post_body: str, +) -> list[PersonMention]: + """Extracts, persists, and returns the `PersonMention`s found in one post. + + Raises whatever `client.extract` raises (e.g. a `NullKeymanExtractionClient` + would raise `RuntimeError`) -- callers should check `client.available` + first, same discipline as every other pluggable channel in this repo. + """ + mentions = client.extract(post_title, post_body) + + for mention in mentions: + person_id = await _upsert_person(conn, mention) + await conn.execute( + "insert into post_person_mention (post_id, person_id) values ($1, $2) on conflict do nothing", + post_id, + person_id, + ) + for organization_name in mention.affiliated_organization_names: + corporate_entity_id = await _resolve_corporate_entity_id(conn, organization_name) + await conn.execute( + """ + insert into person_affiliation (person_id, affiliated_organization_name, affiliated_corporate_entity_id) + values ($1, $2, $3) + on conflict (person_id, affiliated_organization_name) + do update set affiliated_corporate_entity_id = excluded.affiliated_corporate_entity_id + """, + person_id, + organization_name, + corporate_entity_id, + ) + + if mentions: + await persist_edges_for_post(conn, post_id) + + return mentions diff --git a/backend/app/knowledge_graph.py b/backend/app/knowledge_graph.py new file mode 100644 index 000000000..0890b8ec0 --- /dev/null +++ b/backend/app/knowledge_graph.py @@ -0,0 +1,291 @@ +"""Load a Postgres knowledge-graph subgraph and run RWR on it. + +``lineageweave.knowledge_graph`` is the pure math; this module is the +application-layer reader that respects the same ABAC rule as post +endpoints: private posts the account cannot see never become related +nodes, and a Keyman who is only mentioned on such posts is forbidden. +""" + +from __future__ import annotations + +from typing import Any +from uuid import UUID + +import asyncpg + +from lineageweave.knowledge_graph import ( + EDGE_AFFILIATION, + EDGE_CO_MENTION, + EDGE_MENTION, + NODE_CORPORATE_ENTITY, + NODE_PERSON, + NODE_POST, + KnowledgeGraphEdgeSpec, + adjacency_from_edges, + knowledge_graph_edges_for_post, + node_key, + parse_node_key, + random_walk_with_restart, + select_related_nodes, +) + + +def edge_spec_from_row(row: asyncpg.Record) -> KnowledgeGraphEdgeSpec: + return KnowledgeGraphEdgeSpec( + source_node_type_code=row["source_node_type_code"], + source_node_id=str(row["source_node_id"]), + target_node_type_code=row["target_node_type_code"], + target_node_id=str(row["target_node_id"]), + edge_type_code=row["edge_type_code"], + edge_weight=float(row["edge_weight"]), + ) + + +async def fetch_post_keymen(conn: asyncpg.Connection, post_id: str) -> list[dict[str, Any]]: + person_rows = await conn.fetch( + """ + select p.person_id, p.person_name, p.person_side_code, ppm.mention_context + from post_person_mention ppm + join person p on p.person_id = ppm.person_id + where ppm.post_id = $1 + order by p.person_name + """, + post_id, + ) + if not person_rows: + return [] + + affiliation_rows = await conn.fetch( + """ + select person_id, affiliated_organization_name, affiliated_corporate_entity_id, role_title + from person_affiliation + where person_id = any($1::uuid[]) + order by affiliated_organization_name + """, + [row["person_id"] for row in person_rows], + ) + affiliations_by_person: dict[str, list[dict[str, Any]]] = {} + for row in affiliation_rows: + affiliations_by_person.setdefault(str(row["person_id"]), []).append( + { + "organization_name": row["affiliated_organization_name"], + "corporate_entity_id": ( + str(row["affiliated_corporate_entity_id"]) + if row["affiliated_corporate_entity_id"] is not None + else None + ), + "role_title": row["role_title"], + } + ) + + return [ + { + "person_id": str(row["person_id"]), + "person_name": row["person_name"], + "person_side_code": row["person_side_code"], + "mention_context": row["mention_context"], + "affiliations": affiliations_by_person.get(str(row["person_id"]), []), + } + for row in person_rows + ] + + +async def persist_edges_for_post(conn: asyncpg.Connection, post_id: str) -> list[KnowledgeGraphEdgeSpec]: + mention_rows = await conn.fetch( + "select person_id from post_person_mention where post_id = $1", + post_id, + ) + affiliation_rows = await conn.fetch( + """ + select person_id, affiliated_corporate_entity_id + from person_affiliation + where person_id = any($1::uuid[]) + and affiliated_corporate_entity_id is not null + """, + [row["person_id"] for row in mention_rows], + ) + edges = knowledge_graph_edges_for_post( + post_id, + [str(row["person_id"]) for row in mention_rows], + [ + (str(row["person_id"]), str(row["affiliated_corporate_entity_id"])) + for row in affiliation_rows + ], + ) + for edge in edges: + await conn.execute( + """ + insert into knowledge_graph_edge ( + source_node_type_code, source_node_id, + target_node_type_code, target_node_id, + edge_type_code, edge_weight + ) + select $1, $2::uuid, $3, $4::uuid, $5, $6 + where not exists ( + select 1 from knowledge_graph_edge + where source_node_type_code = $1 + and source_node_id = $2::uuid + and target_node_type_code = $3 + and target_node_id = $4::uuid + and edge_type_code = $5 + ) + """, + edge.source_node_type_code, + edge.source_node_id, + edge.target_node_type_code, + edge.target_node_id, + edge.edge_type_code, + edge.edge_weight, + ) + return edges + + +async def person_exists(conn: asyncpg.Connection, person_id: str) -> bool: + try: + UUID(person_id) + except ValueError: + return False + row = await conn.fetchrow("select 1 from person where person_id = $1", person_id) + return row is not None + + +async def visible_mention_post_ids( + conn: asyncpg.Connection, + person_id: str, + can_see_post, +) -> list[str]: + rows = await conn.fetch( + """ + select p.post_id, p.visibility_code, p.corporate_entity_id + from post_person_mention ppm + join post p on p.post_id = ppm.post_id + where ppm.person_id = $1 + """, + person_id, + ) + return [str(row["post_id"]) for row in rows if can_see_post(row)] + + +async def load_visible_subgraph( + conn: asyncpg.Connection, + visible_post_ids: list[str], +) -> list[KnowledgeGraphEdgeSpec]: + if not visible_post_ids: + return [] + person_rows = await conn.fetch( + "select distinct person_id from post_person_mention where post_id = any($1::uuid[])", + visible_post_ids, + ) + person_ids = [row["person_id"] for row in person_rows] + if not person_ids: + return [] + rows = await conn.fetch( + """ + select source_node_type_code, source_node_id, + target_node_type_code, target_node_id, + edge_type_code, edge_weight + from knowledge_graph_edge + where + ( + edge_type_code = $3 + and ( + (source_node_type_code = $4 and source_node_id = any($1::uuid[])) + or (target_node_type_code = $4 and target_node_id = any($1::uuid[])) + ) + ) + or ( + edge_type_code = $5 + and source_node_type_code = $6 + and target_node_type_code = $6 + and source_node_id = any($2::uuid[]) + and target_node_id = any($2::uuid[]) + ) + or ( + edge_type_code = $7 + and ( + (source_node_type_code = $6 and source_node_id = any($2::uuid[])) + or (target_node_type_code = $6 and target_node_id = any($2::uuid[])) + ) + ) + """, + visible_post_ids, + person_ids, + EDGE_MENTION, + NODE_POST, + EDGE_CO_MENTION, + NODE_PERSON, + EDGE_AFFILIATION, + ) + return [edge_spec_from_row(row) for row in rows] + + +async def hydrate_related_nodes( + conn: asyncpg.Connection, + related: list[tuple[str, float]], +) -> list[dict[str, Any]]: + person_ids: list[str] = [] + post_ids: list[str] = [] + corp_ids: list[str] = [] + parsed: list[tuple[str, str, float]] = [] + for key, score in related: + node_type_code, node_id = parse_node_key(key) + parsed.append((node_type_code, node_id, score)) + if node_type_code == NODE_PERSON: + person_ids.append(node_id) + elif node_type_code == NODE_POST: + post_ids.append(node_id) + elif node_type_code == NODE_CORPORATE_ENTITY: + corp_ids.append(node_id) + + people = { + str(row["person_id"]): row + for row in await conn.fetch( + "select person_id, person_name, person_side_code from person where person_id = any($1::uuid[])", + person_ids, + ) + } if person_ids else {} + posts = { + str(row["post_id"]): row + for row in await conn.fetch( + "select post_id, post_title from post where post_id = any($1::uuid[])", + post_ids, + ) + } if post_ids else {} + corps = { + str(row["corporate_entity_id"]): row + for row in await conn.fetch( + "select corporate_entity_id, entity_name from corporate_entity where corporate_entity_id = any($1::uuid[])", + corp_ids, + ) + } if corp_ids else {} + + payload: list[dict[str, Any]] = [] + for node_type_code, node_id, score in parsed: + item: dict[str, Any] = { + "node_id": node_id, + "node_type_code": node_type_code, + "relevance": score, + } + if node_type_code == NODE_PERSON and node_id in people: + item["label"] = people[node_id]["person_name"] + item["person_side_code"] = people[node_id]["person_side_code"] + elif node_type_code == NODE_POST and node_id in posts: + item["label"] = posts[node_id]["post_title"] + elif node_type_code == NODE_CORPORATE_ENTITY and node_id in corps: + item["label"] = corps[node_id]["entity_name"] + else: + continue + payload.append(item) + return payload + + +async def related_for_person( + conn: asyncpg.Connection, + person_id: str, + visible_post_ids: list[str], +) -> list[dict[str, Any]]: + edges = await load_visible_subgraph(conn, visible_post_ids) + start = node_key(NODE_PERSON, person_id) + scores = random_walk_with_restart(adjacency_from_edges(edges), start_node=start) + related = select_related_nodes(scores, start_node=start) + return await hydrate_related_nodes(conn, related) diff --git a/backend/app/main.py b/backend/app/main.py index dfd6874c2..0a3e186d8 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -22,11 +22,24 @@ from fastapi import Depends, FastAPI, HTTPException, status from fastapi.middleware.cors import CORSMiddleware +from lineageweave.keyman_extraction import ( + ContextualOrchestratorKeymanExtractionClient, + NullKeymanExtractionClient, +) + from backend.app.auth import CurrentAccount, get_current_account from backend.app.config import load_settings from backend.app.db import create_pool, get_pool +from backend.app.keyman_ingestion import ingest_post_keymen +from backend.app.knowledge_graph import ( + fetch_post_keymen, + person_exists, + related_for_person, + visible_mention_post_ids, +) _POST_READ = "post_read" +_POST_ADMIN = "post_admin" @asynccontextmanager @@ -43,7 +56,7 @@ async def lifespan(app: FastAPI): app.add_middleware( CORSMiddleware, allow_origins=load_settings().frontend_origins, - allow_methods=["GET"], + allow_methods=["GET", "POST"], allow_headers=["Authorization"], ) @@ -53,6 +66,20 @@ def _require_post_read(account: CurrentAccount) -> None: raise HTTPException(status.HTTP_403_FORBIDDEN, "account lacks the post_read permission") +def _require_post_admin(account: CurrentAccount) -> None: + if not account.has_permission(_POST_ADMIN): + raise HTTPException(status.HTTP_403_FORBIDDEN, "account lacks the post_admin permission") + + +def _keyman_extraction_client(): + settings = load_settings() + if not (settings.orchestrator_base_url and settings.orchestrator_api_key): + return NullKeymanExtractionClient() + return ContextualOrchestratorKeymanExtractionClient( + base_url=settings.orchestrator_base_url, api_key=settings.orchestrator_api_key + ) + + def _can_see_post(account: CurrentAccount, post: asyncpg.Record) -> bool: if post["visibility_code"] == "public": return True @@ -115,3 +142,97 @@ async def read_post( if not _can_see_post(account, row): raise HTTPException(status.HTTP_403_FORBIDDEN, "not authorized to view this post") return {**_serialize_post(row), "post_body": row["post_body"]} + + +async def _load_visible_post( + post_id: str, + account: CurrentAccount, + pool: asyncpg.Pool, +) -> asyncpg.Record: + _require_post_read(account) + async with pool.acquire() as conn: + row = await conn.fetchrow( + "select post_id, post_title, voc_type_code, visibility_code, corporate_entity_id, created_at " + "from post where post_id = $1", + post_id, + ) + if row is None: + raise HTTPException(status.HTTP_404_NOT_FOUND, "post not found") + if not _can_see_post(account, row): + raise HTTPException(status.HTTP_403_FORBIDDEN, "not authorized to view this post") + return row + + +@app.get("/api/posts/{post_id}/keymen") +async def read_post_keymen( + post_id: str, + account: CurrentAccount = Depends(get_current_account), + pool: asyncpg.Pool = Depends(get_pool), +) -> dict[str, Any]: + post = await _load_visible_post(post_id, account, pool) + async with pool.acquire() as conn: + keymen = await fetch_post_keymen(conn, post_id) + return {"post_id": str(post["post_id"]), "keymen": keymen} + + +@app.get("/api/keymen/{person_id}/related") +async def read_related_keymen( + person_id: str, + account: CurrentAccount = Depends(get_current_account), + pool: asyncpg.Pool = Depends(get_pool), +) -> dict[str, Any]: + _require_post_read(account) + async with pool.acquire() as conn: + if not await person_exists(conn, person_id): + raise HTTPException(status.HTTP_404_NOT_FOUND, "person not found") + visible_post_ids = await visible_mention_post_ids(conn, person_id, lambda row: _can_see_post(account, row)) + if not visible_post_ids: + raise HTTPException(status.HTTP_403_FORBIDDEN, "not authorized to view this person") + person = await conn.fetchrow( + "select person_id, person_name, person_side_code from person where person_id = $1", + person_id, + ) + related = await related_for_person(conn, person_id, visible_post_ids) + return { + "person_id": str(person["person_id"]), + "person_name": person["person_name"], + "person_side_code": person["person_side_code"], + "related": related, + } + + +@app.post("/api/posts/{post_id}/extract-keymen") +async def extract_post_keymen( + post_id: str, + account: CurrentAccount = Depends(get_current_account), + pool: asyncpg.Pool = Depends(get_pool), +) -> dict[str, Any]: + """Runs Keyman extraction over a post's own title+body and persists the + result (person / person_affiliation / post_person_mention / + knowledge_graph_edge). Gated by post_admin, not post_read: this is a + write action with a real LLM-call cost, not a read. + """ + _require_post_admin(account) + post = await _load_visible_post(post_id, account, pool) + client = _keyman_extraction_client() + if not client.available: + raise HTTPException( + status.HTTP_503_SERVICE_UNAVAILABLE, + "Keyman extraction is unavailable: set ORCHESTRATOR_BASE_URL / ORCHESTRATOR_API_KEY", + ) + async with pool.acquire() as conn: + body_row = await conn.fetchrow("select post_body from post where post_id = $1", post_id) + async with conn.transaction(): + mentions = await ingest_post_keymen(conn, client, post_id, post["post_title"], body_row["post_body"]) + return { + "post_id": str(post["post_id"]), + "extracted_count": len(mentions), + "mentions": [ + { + "person_name": mention.person_name, + "person_side_code": mention.person_side_code, + "affiliated_organization_names": list(mention.affiliated_organization_names), + } + for mention in mentions + ], + } diff --git a/backend/tests/test_api.py b/backend/tests/test_api.py index f1704df05..7b124c46d 100644 --- a/backend/tests/test_api.py +++ b/backend/tests/test_api.py @@ -104,7 +104,15 @@ def seeded_db(demo_analyst_token): "('post_visibility', 'public', 'Public'), " "('post_visibility', 'private', 'Private'), " "('voc_type', 'voc', 'Voice of Customer'), " - "('permission', 'post_read', 'Read posts')" + "('permission', 'post_read', 'Read posts'), " + "('person_side', 'our_side', 'Our side'), " + "('person_side', 'counterparty', 'Counterparty'), " + "('node_type', 'node_person', 'Person'), " + "('node_type', 'node_corporate_entity', 'Corporate entity'), " + "('node_type', 'node_post', 'Post'), " + "('edge_type', 'edge_mention', 'Mentioned in'), " + "('edge_type', 'edge_affiliation', 'Affiliated with'), " + "('edge_type', 'edge_co_mention', 'Co-mentioned')" ) cur.execute( "insert into corporate_entity (corporate_entity_code, entity_name, entity_level_code) " @@ -151,6 +159,82 @@ def _insert_post(title: str, corporate_entity_id, visibility_code: str) -> str: public_post_id = _insert_post("Public post", other_corp_id, "public") own_private_post_id = _insert_post("Own-corp private post", own_corp_id, "private") other_private_post_id = _insert_post("Other-corp private post", other_corp_id, "private") + + cur.execute( + "insert into person (person_name, person_side_code) values " + "('Ada West', 'our_side') returning person_id" + ) + our_person_id = str(cur.fetchone()[0]) + cur.execute( + "insert into person (person_name, person_side_code) values " + "('Priya Nair', 'counterparty') returning person_id" + ) + counterpart_person_id = str(cur.fetchone()[0]) + cur.execute( + "insert into person (person_name, person_side_code) values " + "('Other Corp Only', 'counterparty') returning person_id" + ) + hidden_person_id = str(cur.fetchone()[0]) + + cur.execute( + "insert into person_affiliation (person_id, affiliated_organization_name, affiliated_corporate_entity_id) " + "values (%s, 'Test Corp', %s)", + (our_person_id, own_corp_id), + ) + cur.execute( + "insert into person_affiliation (person_id, affiliated_organization_name) " + "values (%s, 'Northridge Grid'), (%s, 'Northridge Holdings')", + (counterpart_person_id, counterpart_person_id), + ) + + cur.execute( + "insert into post_person_mention (post_id, person_id) values " + "(%s, %s), (%s, %s), (%s, %s), (%s, %s)", + ( + own_private_post_id, + our_person_id, + own_private_post_id, + counterpart_person_id, + public_post_id, + our_person_id, + other_private_post_id, + hidden_person_id, + ), + ) + + from lineageweave.knowledge_graph import knowledge_graph_edges_for_post + + seen_edges: set[tuple[str, str, str, str, str]] = set() + for post_id, person_ids, affiliations in ( + (own_private_post_id, [our_person_id, counterpart_person_id], [(our_person_id, str(own_corp_id))]), + (public_post_id, [our_person_id], [(our_person_id, str(own_corp_id))]), + (other_private_post_id, [hidden_person_id], []), + ): + for edge in knowledge_graph_edges_for_post(post_id, person_ids, affiliations): + key = ( + edge.source_node_type_code, + edge.source_node_id, + edge.target_node_type_code, + edge.target_node_id, + edge.edge_type_code, + ) + if key in seen_edges: + continue + seen_edges.add(key) + cur.execute( + "insert into knowledge_graph_edge (" + "source_node_type_code, source_node_id, target_node_type_code, " + "target_node_id, edge_type_code, edge_weight" + ") values (%s, %s, %s, %s, %s, %s)", + ( + edge.source_node_type_code, + edge.source_node_id, + edge.target_node_type_code, + edge.target_node_id, + edge.edge_type_code, + edge.edge_weight, + ), + ) conn.commit() yield { @@ -158,6 +242,9 @@ def _insert_post(title: str, corporate_entity_id, visibility_code: str) -> str: "public_post_id": public_post_id, "own_private_post_id": own_private_post_id, "other_private_post_id": other_private_post_id, + "our_person_id": our_person_id, + "counterpart_person_id": counterpart_person_id, + "hidden_person_id": hidden_person_id, } finally: conn.close() @@ -226,3 +313,132 @@ def test_forged_token_is_rejected(client) -> None: forged = jwt.encode({"sub": "not-a-real-subject", "iss": f"{_KEYCLOAK_BASE_URL}/realms/{_REALM}"}, key="wrong-key", algorithm="HS256") response = client.get("/api/posts", headers={"Authorization": f"Bearer {forged}"}) assert response.status_code == 401 + + +def test_own_corp_post_keymen_are_readable(client, demo_analyst_token, seeded_db) -> None: + response = client.get( + f"/api/posts/{seeded_db['own_private_post_id']}/keymen", + headers={"Authorization": f"Bearer {demo_analyst_token}"}, + ) + assert response.status_code == 200 + names = {person["person_name"]: person for person in response.json()["keymen"]} + assert set(names) == {"Ada West", "Priya Nair"} + assert names["Ada West"]["person_side_code"] == "our_side" + assert names["Priya Nair"]["person_side_code"] == "counterparty" + assert {aff["organization_name"] for aff in names["Priya Nair"]["affiliations"]} == { + "Northridge Grid", + "Northridge Holdings", + } + + +def test_other_corp_private_post_keymen_are_forbidden(client, demo_analyst_token, seeded_db) -> None: + response = client.get( + f"/api/posts/{seeded_db['other_private_post_id']}/keymen", + headers={"Authorization": f"Bearer {demo_analyst_token}"}, + ) + assert response.status_code == 403 + + +def test_related_keymen_use_rwr_and_hide_invisible_posts(client, demo_analyst_token, seeded_db) -> None: + response = client.get( + f"/api/keymen/{seeded_db['our_person_id']}/related", + headers={"Authorization": f"Bearer {demo_analyst_token}"}, + ) + assert response.status_code == 200 + body = response.json() + assert body["person_name"] == "Ada West" + related_ids = {node["node_id"] for node in body["related"]} + assert seeded_db["counterpart_person_id"] in related_ids + assert seeded_db["own_private_post_id"] in related_ids + assert seeded_db["hidden_person_id"] not in related_ids + assert seeded_db["other_private_post_id"] not in related_ids + assert all(node["relevance"] > 0 for node in body["related"]) + + +def test_keyman_only_on_other_corp_private_post_is_forbidden(client, demo_analyst_token, seeded_db) -> None: + response = client.get( + f"/api/keymen/{seeded_db['hidden_person_id']}/related", + headers={"Authorization": f"Bearer {demo_analyst_token}"}, + ) + assert response.status_code == 403 + + +def test_extract_keymen_requires_post_admin(client, demo_analyst_token, seeded_db) -> None: + response = client.post( + f"/api/posts/{seeded_db['own_private_post_id']}/extract-keymen", + headers={"Authorization": f"Bearer {demo_analyst_token}"}, + ) + assert response.status_code == 403 + + +_ORCHESTRATOR_BASE_URL = os.environ.get("LINEAGEWEAVE_TEST_ORCHESTRATOR_BASE_URL") +_ORCHESTRATOR_API_KEY = os.environ.get("LINEAGEWEAVE_TEST_ORCHESTRATOR_API_KEY") + + +@pytest.mark.skipif( + not (_ORCHESTRATOR_BASE_URL and _ORCHESTRATOR_API_KEY), + reason="set LINEAGEWEAVE_TEST_ORCHESTRATOR_BASE_URL and LINEAGEWEAVE_TEST_ORCHESTRATOR_API_KEY to run", +) +def test_extract_keymen_persists_a_real_llm_extraction(client, demo_analyst_token, seeded_db) -> None: + """The full write path, end to end: a real LLM call through a live + contextual-orchestrator, persisted to Postgres, then read back through + the read endpoints -- not a mocked extraction client. + """ + os.environ["ORCHESTRATOR_BASE_URL"] = _ORCHESTRATOR_BASE_URL + os.environ["ORCHESTRATOR_API_KEY"] = _ORCHESTRATOR_API_KEY + + from lineageweave.fixtures import ambiguous_keyman_post + + title, body = ambiguous_keyman_post() + + admin_conn = psycopg2.connect(seeded_db["dsn"]) + admin_conn.autocommit = True + try: + with admin_conn.cursor() as cur: + cur.execute( + "insert into common_lookup_value (lookup_category, lookup_code, lookup_label) " + "values ('permission', 'post_admin', 'Administer posts') on conflict (lookup_code) do nothing" + ) + cur.execute("select access_role_id from account_role_assignment limit 1") + role_id = cur.fetchone()[0] + cur.execute( + "insert into role_permission (access_role_id, permission_code) values (%s, 'post_admin') " + "on conflict do nothing", + (role_id,), + ) + cur.execute( + "insert into post (author_account_id, corporate_entity_id, post_title, post_body, voc_type_code, visibility_code) " + "select author_account_id, corporate_entity_id, %s, %s, 'voc', 'public' " + "from post where post_id = %s " + "returning post_id", + (title, body, seeded_db["own_private_post_id"]), + ) + new_post_id = str(cur.fetchone()[0]) + finally: + admin_conn.close() + + response = client.post( + f"/api/posts/{new_post_id}/extract-keymen", + headers={"Authorization": f"Bearer {demo_analyst_token}"}, + ) + assert response.status_code == 200, response.text + body_json = response.json() + assert body_json["extracted_count"] >= 2 + names = {mention["person_name"] for mention in body_json["mentions"]} + assert any("Jordan" in name for name in names) + assert any("Priya" in name for name in names) + + keymen_response = client.get( + f"/api/posts/{new_post_id}/keymen", headers={"Authorization": f"Bearer {demo_analyst_token}"} + ) + assert keymen_response.status_code == 200 + persisted_names = {person["person_name"] for person in keymen_response.json()["keymen"]} + assert persisted_names == names + + +def test_unknown_keyman_is_not_found(client, demo_analyst_token) -> None: + response = client.get( + f"/api/keymen/{uuid.uuid4()}/related", + headers={"Authorization": f"Bearer {demo_analyst_token}"}, + ) + assert response.status_code == 404 diff --git a/docs/lineage-bi-research-notes.md b/docs/lineage-bi-research-notes.md index 89544f26f..fb4297c37 100644 --- a/docs/lineage-bi-research-notes.md +++ b/docs/lineage-bi-research-notes.md @@ -237,6 +237,21 @@ Additional context on the Fugu / Conductor / TRINITY test-time-compute-allocatio [contextual-orchestrator's own literature register](https://github.com/ContextualWisdomLab/contextual-orchestrator/blob/main/docs/architecture.md) rather than duplicated here, so the two repos do not drift out of sync. +## Keyman extraction (Phase 2) + +`lineageweave/keyman_extraction.py` treats a named person in a post as an +ACE-style entity mention (Doddington et al., 2004): the surface string is +not yet a resolved person node, and a mention whose side cannot be +classified into the closed `{our_side, counterparty}` set is dropped +rather than guessed. N:N organization attachments are slot-filling on +that mention (a person may have zero, one, or several affiliations in +the same post), not a second independent NER pass. The live client +calls contextual-orchestrator (`mode="route"`) rather than a raw LLM +API so reasoning-effort allocation stays centralized with the +adjudication channel. Proven for real during development against +`fixtures.ambiguous_keyman_post` when orchestrator credentials are set; +the default suite asserts the parser and the never-fake null client. + ## Knowledge Graph traversal (Phase 2) `lineageweave/knowledge_graph.py`'s `random_walk_with_restart` implements diff --git a/lineageweave/__init__.py b/lineageweave/__init__.py index feb458b3a..c2a1d7a1f 100644 --- a/lineageweave/__init__.py +++ b/lineageweave/__init__.py @@ -20,4 +20,4 @@ "select_related_nodes", ] -__version__ = "0.8.0" +__version__ = "0.9.0" diff --git a/lineageweave/fixtures.py b/lineageweave/fixtures.py index b0f34d8aa..c7fe59111 100644 --- a/lineageweave/fixtures.py +++ b/lineageweave/fixtures.py @@ -38,3 +38,24 @@ def sample_records() -> list[Record]: Record("rec-102", "B-200", "Specification revision requested", _day(8), "proj-beta"), Record("rec-103", "B-200", "Revised specification approved", _day(15), "proj-beta"), ] + + +def ambiguous_keyman_post() -> tuple[str, str]: + """A synthetic post that is not a trivially-templated 'Alice of Acme' + list. Side and affiliation have to be read out of running prose: + one of our people, one dual-hatted counterparty, one internal + counsel mentioned only by role, and an organization that sent + nobody (so it must not be invented as a person). + """ + title = "Follow-up after the Northridge transformer bid workshop" + body = ( + "Jordan Hale walked our sales team through the revised bid timeline " + "and asked Priya Nair (who sits on both Northridge Grid and its " + "parent, Northridge Holdings) to confirm the inspection window. " + "Jordan also looped in our legal counsel, Sam Okonkwo, because " + "Priya's dual role at the holding company is what made the " + "warranty language messy last quarter. No one from Westfield " + "Power attended -- they sent a note saying they would review the " + "minutes later." + ) + return title, body diff --git a/lineageweave/keyman_extraction.py b/lineageweave/keyman_extraction.py new file mode 100644 index 000000000..2d9db3800 --- /dev/null +++ b/lineageweave/keyman_extraction.py @@ -0,0 +1,154 @@ +"""Pluggable Keyman extraction: which people does a post mention, are they +"our side" or a counterparty, and which organizations are they affiliated +with (a person may have N affiliations -- an internal Keyman can span +multiple group companies, a counterparty Keyman can span multiple external +orgs, per the product requirement). + +The default :class:`NullKeymanExtractionClient` makes the channel +unavailable, same never-fake-a-missing-signal discipline as +``embedding_client``/``adjudication_client``/``image_content``. +:class:`ContextualOrchestratorKeymanExtractionClient` calls a running +contextual-orchestrator instance -- never a raw LLM API directly, per +AGENTS.md -- because Keyman identification is a structured-extraction task +that benefits from the orchestrator's reasoning-effort allocation, not a +single confidence number, so it uses ``mode="route"`` (one worker call) at +a ``"medium"`` reasoning effort by default rather than ``verify``'s +worker-plus-checker pattern, which is reserved for adjudication's binary +judgment calls. +""" + +from __future__ import annotations + +import json +import re +from dataclasses import dataclass, field +from typing import Protocol + +from .http_client import post_json + +OUR_SIDE = "our_side" +COUNTERPARTY = "counterparty" +_VALID_SIDES = frozenset({OUR_SIDE, COUNTERPARTY}) + + +@dataclass(frozen=True) +class PersonMention: + """One person the extractor found in a post's text. + + ``affiliated_organization_names`` may be empty (mentioned without a + stated affiliation) or contain more than one name (the N:N case the + product requirement describes). + """ + + person_name: str + person_side_code: str + affiliated_organization_names: tuple[str, ...] = field(default_factory=tuple) + + +class KeymanExtractionClient(Protocol): + """Extracts person mentions from a post's title + body.""" + + available: bool + + def extract(self, post_title: str, post_body: str) -> list[PersonMention]: ... + + +class NullKeymanExtractionClient: + """No LLM orchestrator configured -- Keyman extraction is unavailable.""" + + available = False + + def extract(self, post_title: str, post_body: str) -> list[PersonMention]: + raise RuntimeError("NullKeymanExtractionClient cannot extract; check .available first") + + +_EXTRACTION_PROMPT_TEMPLATE = """\ +Read the post below and list every named person it mentions. For each +person, classify which side they are on and list every organization they +are affiliated with according to the text (a person may belong to more +than one organization, or none if the text does not say). + +Reply with ONLY a JSON array (no markdown fences, no prose), where each +element has exactly these fields: + "name": the person's name as written in the text + "side": either "our_side" (the post author's own organization/group) or + "counterparty" (an external customer, partner, competitor, or + other outside organization) + "affiliations": a JSON array of organization name strings (can be empty) + +If no people are named, reply with an empty JSON array: [] + +Post title: {title} +Post body: {body} +""" + +_CODE_FENCE_PATTERN = re.compile(r"```(?:json)?\s*(.*?)\s*```", re.DOTALL) + + +def _strip_code_fence(content: str) -> str: + match = _CODE_FENCE_PATTERN.search(content) + return match.group(1) if match else content + + +def parse_keyman_response(content: str) -> list[PersonMention]: + """Parses the LLM's JSON array response into `PersonMention`s. + + Entries missing a name, or whose "side" isn't one of the two valid + codes, are skipped rather than guessed at -- a wrong Keyman-side + classification is worse than a dropped mention, since Phase 2's + Knowledge Graph edges are typed by this code. + """ + try: + parsed = json.loads(_strip_code_fence(content).strip()) + except json.JSONDecodeError: + return [] + if not isinstance(parsed, list): + return [] + + mentions: list[PersonMention] = [] + for entry in parsed: + if not isinstance(entry, dict): + continue + name = entry.get("name") + side = entry.get("side") + if not isinstance(name, str) or not name.strip(): + continue + if side not in _VALID_SIDES: + continue + affiliations_raw = entry.get("affiliations") or [] + if not isinstance(affiliations_raw, list): + affiliations_raw = [] + affiliations = tuple(a.strip() for a in affiliations_raw if isinstance(a, str) and a.strip()) + mentions.append( + PersonMention(person_name=name.strip(), person_side_code=side, affiliated_organization_names=affiliations) + ) + return mentions + + +class ContextualOrchestratorKeymanExtractionClient: + """Calls ``POST {base_url}/v1/chat/completions`` with ``mode="route"``.""" + + available = True + + def __init__( + self, base_url: str, api_key: str, *, reasoning_effort: str = "medium", timeout: float = 60.0 + ) -> None: + self._base_url = base_url.rstrip("/") + self._api_key = api_key + self._reasoning_effort = reasoning_effort + self._timeout = timeout + + def extract(self, post_title: str, post_body: str) -> list[PersonMention]: + prompt = _EXTRACTION_PROMPT_TEMPLATE.format(title=post_title, body=post_body) + body = post_json( + f"{self._base_url}/v1/chat/completions", + { + "messages": [{"role": "user", "content": prompt}], + "mode": "route", + "reasoning_effort": self._reasoning_effort, + }, + headers={"authorization": f"Bearer {self._api_key}"}, + timeout=self._timeout, + ) + content = body["choices"][0]["message"]["content"] + return parse_keyman_response(content) diff --git a/lineageweave/knowledge_graph.py b/lineageweave/knowledge_graph.py index 7b7b4aca4..501703968 100644 --- a/lineageweave/knowledge_graph.py +++ b/lineageweave/knowledge_graph.py @@ -9,14 +9,19 @@ `depth=2` BFS cannot express that; a relevance-score cutoff can, because it adapts to each node's own local graph structure. -This module is pure graph math -- no Postgres, no knowledge_graph_edge -schema awareness. See backend/app/knowledge_graph.py for the part that -loads a Postgres subgraph into the adjacency shape this module expects. +This module is pure graph math plus the typed edge-spec builder that +turns persons/affiliations/posts into ``knowledge_graph_edge`` rows. +It does not talk to Postgres. See backend/app/knowledge_graph.py for +the part that loads a Postgres subgraph into the adjacency shape this +module expects. """ from __future__ import annotations from collections import defaultdict +from dataclasses import dataclass +from itertools import combinations +from typing import Sequence Adjacency = dict[str, dict[str, float]] @@ -99,3 +104,116 @@ def select_related_nodes( return [] threshold = top_score * min_relevance_ratio return [(node, score) for node, score in candidates if score >= threshold][:max_nodes] + + +# Lookup codes written into knowledge_graph_edge / common_lookup_value. +# Prefixed so they stay unique across every lookup_category (the schema +# enforces unique(lookup_code) globally -- see migrations/0001). +NODE_PERSON = "node_person" +NODE_CORPORATE_ENTITY = "node_corporate_entity" +NODE_POST = "node_post" +EDGE_MENTION = "edge_mention" +EDGE_AFFILIATION = "edge_affiliation" +EDGE_CO_MENTION = "edge_co_mention" + + +@dataclass(frozen=True) +class KnowledgeGraphEdgeSpec: + """One typed edge the application layer persists to knowledge_graph_edge. + + Node ids are opaque strings (UUIDs in the product schema). The + polymorphic id columns have no FK -- this spec is the application-layer + contract that a writer must have already confirmed the endpoints exist. + """ + + source_node_type_code: str + source_node_id: str + target_node_type_code: str + target_node_id: str + edge_type_code: str + edge_weight: float = 1.0 + + +def node_key(node_type_code: str, node_id: str) -> str: + return f"{node_type_code}:{node_id}" + + +def parse_node_key(key: str) -> tuple[str, str]: + node_type_code, node_id = key.split(":", 1) + return node_type_code, node_id + + +def knowledge_graph_edges_for_post( + post_id: str, + person_ids: Sequence[str], + person_corporate_entity_ids: Sequence[tuple[str, str]] = (), +) -> list[KnowledgeGraphEdgeSpec]: + """Populate the three Phase 2 edge kinds for one post. + + - person <-> post (``edge_mention``) for every mentioned person + - person <-> corporate_entity (``edge_affiliation``) for every + affiliation that resolved to a real ``corporate_entity`` row + - person <-> person (``edge_co_mention``) for every unordered pair of + people named in the same post + + Affiliation names that did not resolve to a ``corporate_entity`` are + stored on ``person_affiliation`` but do not become graph edges -- a + free-text org with no node id cannot be a knowledge_graph_edge + endpoint. Directed storage is canonical (person -> post/org, and + lexicographic person-id order for co-mentions); loaders treat the + graph as undirected. + """ + unique_person_ids = list(dict.fromkeys(person_ids)) + edges: list[KnowledgeGraphEdgeSpec] = [] + + for person_id in unique_person_ids: + edges.append( + KnowledgeGraphEdgeSpec( + source_node_type_code=NODE_PERSON, + source_node_id=person_id, + target_node_type_code=NODE_POST, + target_node_id=post_id, + edge_type_code=EDGE_MENTION, + ) + ) + + seen_affiliations: set[tuple[str, str]] = set() + for person_id, corporate_entity_id in person_corporate_entity_ids: + pair = (person_id, corporate_entity_id) + if pair in seen_affiliations: + continue + seen_affiliations.add(pair) + edges.append( + KnowledgeGraphEdgeSpec( + source_node_type_code=NODE_PERSON, + source_node_id=person_id, + target_node_type_code=NODE_CORPORATE_ENTITY, + target_node_id=corporate_entity_id, + edge_type_code=EDGE_AFFILIATION, + ) + ) + + for left, right in combinations(unique_person_ids, 2): + source_id, target_id = (left, right) if left < right else (right, left) + edges.append( + KnowledgeGraphEdgeSpec( + source_node_type_code=NODE_PERSON, + source_node_id=source_id, + target_node_type_code=NODE_PERSON, + target_node_id=target_id, + edge_type_code=EDGE_CO_MENTION, + ) + ) + + return edges + + +def adjacency_from_edges(edges: Sequence[KnowledgeGraphEdgeSpec]) -> Adjacency: + """Undirected weighted adjacency for RWR, keyed as ``type:id``.""" + adjacency: Adjacency = {} + for edge in edges: + source = node_key(edge.source_node_type_code, edge.source_node_id) + target = node_key(edge.target_node_type_code, edge.target_node_id) + adjacency.setdefault(source, {})[target] = float(edge.edge_weight) + adjacency.setdefault(target, {})[source] = float(edge.edge_weight) + return adjacency diff --git a/pyproject.toml b/pyproject.toml index 3fd8b37d6..aea6038d2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "lineageweave" -version = "0.8.0" +version = "0.9.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 dc7401ba2..835545599 100644 --- a/scripts/seed_demo_data.py +++ b/scripts/seed_demo_data.py @@ -79,7 +79,15 @@ def seed(postgres_dsn: str, subjects: dict[str, str]) -> None: ('voc_type', 'voc', 'Voice of Customer', 0), ('voc_type', 'vom', 'Voice of Market', 1), ('permission', 'post_read', 'Read posts', 0), - ('permission', 'post_admin', 'Administer posts', 1) + ('permission', 'post_admin', 'Administer posts', 1), + ('person_side', 'our_side', 'Our side', 0), + ('person_side', 'counterparty', 'Counterparty', 1), + ('node_type', 'node_person', 'Person', 0), + ('node_type', 'node_corporate_entity', 'Corporate entity', 1), + ('node_type', 'node_post', 'Post', 2), + ('edge_type', 'edge_mention', 'Mentioned in', 0), + ('edge_type', 'edge_affiliation', 'Affiliated with', 1), + ('edge_type', 'edge_co_mention', 'Co-mentioned', 2) on conflict (lookup_code) do nothing """ ) @@ -160,6 +168,52 @@ def seed(postgres_dsn: str, subjects: dict[str, str]) -> None: (account_ids["demo.admin"], corporate_entity_id, process_units["DEMO-PU-HQ"]), ) + cur.execute("select post_id from post where post_title = 'Demo public post'") + demo_public_post_id = cur.fetchone()[0] + cur.execute("select person_id from person where person_name = 'Ada West'") + if cur.fetchone() is None: + from lineageweave.knowledge_graph import knowledge_graph_edges_for_post + + cur.execute( + "insert into person (person_name, person_side_code) values " + "('Ada West', 'our_side'), ('Priya Nair', 'counterparty') " + "returning person_name, person_id" + ) + people = dict(cur.fetchall()) + cur.execute( + "insert into person_affiliation (person_id, affiliated_organization_name, affiliated_corporate_entity_id) " + "values (%s, 'Demo Corp', %s)", + (people["Ada West"], corporate_entity_id), + ) + cur.execute( + "insert into person_affiliation (person_id, affiliated_organization_name) values " + "(%s, 'Northridge Grid'), (%s, 'Northridge Holdings')", + (people["Priya Nair"], people["Priya Nair"]), + ) + cur.execute( + "insert into post_person_mention (post_id, person_id) values (%s, %s), (%s, %s)", + (demo_public_post_id, people["Ada West"], demo_public_post_id, people["Priya Nair"]), + ) + for edge in knowledge_graph_edges_for_post( + str(demo_public_post_id), + [str(people["Ada West"]), str(people["Priya Nair"])], + [(str(people["Ada West"]), str(corporate_entity_id))], + ): + cur.execute( + "insert into knowledge_graph_edge (" + "source_node_type_code, source_node_id, target_node_type_code, " + "target_node_id, edge_type_code, edge_weight" + ") values (%s, %s, %s, %s, %s, %s)", + ( + edge.source_node_type_code, + edge.source_node_id, + edge.target_node_type_code, + edge.target_node_id, + edge.edge_type_code, + edge.edge_weight, + ), + ) + conn.commit() finally: conn.close() diff --git a/tests/test_keyman_extraction.py b/tests/test_keyman_extraction.py new file mode 100644 index 000000000..a8a1c9e48 --- /dev/null +++ b/tests/test_keyman_extraction.py @@ -0,0 +1,104 @@ +"""Tests for lineageweave.keyman_extraction. + +parse_keyman_response's tests need no live provider -- they exercise the +JSON-shape validation directly. The real-provider test is skipped unless +LINEAGEWEAVE_TEST_ORCHESTRATOR_BASE_URL/_API_KEY are set (same env vars +ContextualOrchestratorAdjudicationClient's real-provider test uses), and +runs against fixtures.ambiguous_keyman_post() -- a synthetic post +deliberately written as running prose, not a templated "Alice of Acme" +list, so a correct extraction has to actually read the text rather than +pattern-match an obvious structure. +""" + +from __future__ import annotations + +import os + +import pytest + +from lineageweave.fixtures import ambiguous_keyman_post +from lineageweave.keyman_extraction import ( + COUNTERPARTY, + OUR_SIDE, + ContextualOrchestratorKeymanExtractionClient, + parse_keyman_response, +) + + +def test_parses_a_well_formed_json_array() -> None: + content = ( + '[{"name": "Alex Kim", "side": "our_side", "affiliations": []}, ' + '{"name": "Sam Lee", "side": "counterparty", "affiliations": ["Acme Corp", "Acme Holdings"]}]' + ) + mentions = parse_keyman_response(content) + assert len(mentions) == 2 + assert mentions[0].person_name == "Alex Kim" + assert mentions[0].person_side_code == OUR_SIDE + assert mentions[0].affiliated_organization_names == () + assert mentions[1].person_side_code == COUNTERPARTY + assert mentions[1].affiliated_organization_names == ("Acme Corp", "Acme Holdings") + + +def test_strips_a_markdown_code_fence() -> None: + content = '```json\n[{"name": "Jo Park", "side": "our_side", "affiliations": []}]\n```' + mentions = parse_keyman_response(content) + assert len(mentions) == 1 + assert mentions[0].person_name == "Jo Park" + + +def test_empty_array_is_no_mentions() -> None: + assert parse_keyman_response("[]") == [] + + +def test_entry_missing_name_is_skipped() -> None: + content = '[{"side": "our_side", "affiliations": []}]' + assert parse_keyman_response(content) == [] + + +def test_entry_with_invalid_side_is_skipped() -> None: + content = '[{"name": "Ambiguous Person", "side": "unknown", "affiliations": []}]' + assert parse_keyman_response(content) == [] + + +def test_invalid_json_returns_empty_list() -> None: + assert parse_keyman_response("not json at all") == [] + + +def test_non_list_top_level_returns_empty_list() -> None: + assert parse_keyman_response('{"name": "not a list"}') == [] + + +_ORCHESTRATOR_BASE_URL = os.environ.get("LINEAGEWEAVE_TEST_ORCHESTRATOR_BASE_URL") +_ORCHESTRATOR_API_KEY = os.environ.get("LINEAGEWEAVE_TEST_ORCHESTRATOR_API_KEY") + + +@pytest.mark.skipif( + not (_ORCHESTRATOR_BASE_URL and _ORCHESTRATOR_API_KEY), + reason="set LINEAGEWEAVE_TEST_ORCHESTRATOR_BASE_URL and LINEAGEWEAVE_TEST_ORCHESTRATOR_API_KEY to run", +) +def test_contextual_orchestrator_extracts_keymen_from_an_ambiguous_post() -> None: + """A real LLM call, with real assertions grounded in the fixture's + actual content -- not just 'the call didn't crash'. The fixture names + four people; a correct extraction must find the two unambiguously + described ones, classify Jordan as our-side and Priya as counterparty, + and must not invent a person for Westfield Power (an org that sent no + one). + """ + client = ContextualOrchestratorKeymanExtractionClient( + base_url=_ORCHESTRATOR_BASE_URL, api_key=_ORCHESTRATOR_API_KEY + ) + title, body = ambiguous_keyman_post() + + mentions = client.extract(title, body) + + names = {mention.person_name for mention in mentions} + assert any("Jordan" in name for name in names) + assert any("Priya" in name for name in names) + assert not any("Westfield" in name for name in names) + + by_name = {mention.person_name: mention for mention in mentions} + jordan = next(m for name, m in by_name.items() if "Jordan" in name) + priya = next(m for name, m in by_name.items() if "Priya" in name) + assert jordan.person_side_code == OUR_SIDE + assert priya.person_side_code == COUNTERPARTY + assert len(priya.affiliated_organization_names) >= 2 diff --git a/tests/test_knowledge_graph.py b/tests/test_knowledge_graph.py index 01c12ee3f..d4cd59978 100644 --- a/tests/test_knowledge_graph.py +++ b/tests/test_knowledge_graph.py @@ -18,7 +18,18 @@ import pytest -from lineageweave.knowledge_graph import random_walk_with_restart, select_related_nodes +from lineageweave.knowledge_graph import ( + EDGE_AFFILIATION, + EDGE_CO_MENTION, + EDGE_MENTION, + NODE_CORPORATE_ENTITY, + NODE_PERSON, + NODE_POST, + adjacency_from_edges, + knowledge_graph_edges_for_post, + random_walk_with_restart, + select_related_nodes, +) @pytest.fixture @@ -68,3 +79,42 @@ def test_start_node_absent_from_graph_returns_only_itself() -> None: def test_select_related_nodes_empty_graph_returns_empty_list() -> None: assert select_related_nodes({"only": 1.0}, start_node="only") == [] + + +def test_knowledge_graph_edges_for_post_covers_mention_affiliation_and_co_mention() -> None: + edges = knowledge_graph_edges_for_post( + post_id="post-1", + person_ids=["person-b", "person-a", "person-b"], + person_corporate_entity_ids=[("person-a", "corp-1"), ("person-a", "corp-1")], + ) + kinds = {(edge.edge_type_code, edge.source_node_id, edge.target_node_id) for edge in edges} + assert kinds == { + (EDGE_MENTION, "person-b", "post-1"), + (EDGE_MENTION, "person-a", "post-1"), + (EDGE_AFFILIATION, "person-a", "corp-1"), + (EDGE_CO_MENTION, "person-a", "person-b"), + } + assert all(edge.source_node_type_code == NODE_PERSON for edge in edges) + assert {edge.target_node_type_code for edge in edges} == { + NODE_POST, + NODE_CORPORATE_ENTITY, + NODE_PERSON, + } + + +def test_rwr_from_a_keyman_reaches_co_mentioned_person_and_affiliated_org() -> None: + edges = knowledge_graph_edges_for_post( + post_id="post-1", + person_ids=["person-a", "person-b"], + person_corporate_entity_ids=[("person-a", "corp-1")], + ) + scores = random_walk_with_restart( + adjacency_from_edges(edges), start_node=f"{NODE_PERSON}:person-a" + ) + related = dict( + select_related_nodes(scores, start_node=f"{NODE_PERSON}:person-a", min_relevance_ratio=0.05) + ) + assert f"{NODE_PERSON}:person-b" in related + assert f"{NODE_POST}:post-1" in related + assert f"{NODE_CORPORATE_ENTITY}:corp-1" in related + assert related[f"{NODE_PERSON}:person-b"] > 0 diff --git a/tests/test_real_provider_integration.py b/tests/test_real_provider_integration.py index cb8e33826..c5dda7bb6 100644 --- a/tests/test_real_provider_integration.py +++ b/tests/test_real_provider_integration.py @@ -21,7 +21,13 @@ chunked_max_similarity, cosine_similarity, ) +from lineageweave.fixtures import ambiguous_keyman_post from lineageweave.image_content import OpenAiCompatibleVisionClient +from lineageweave.keyman_extraction import ( + COUNTERPARTY, + OUR_SIDE, + ContextualOrchestratorKeymanExtractionClient, +) _EMBEDDING_BASE_URL = os.environ.get("LINEAGEWEAVE_TEST_EMBEDDING_BASE_URL") _EMBEDDING_API_KEY = os.environ.get("LINEAGEWEAVE_TEST_EMBEDDING_API_KEY") @@ -139,3 +145,40 @@ def test_contextual_orchestrator_adjudication_client_returns_a_real_confidence() ) assert 0.0 <= confidence <= 1.0 + + +@pytest.mark.skipif( + not (_ORCHESTRATOR_BASE_URL and _ORCHESTRATOR_API_KEY), + reason=( + "set LINEAGEWEAVE_TEST_ORCHESTRATOR_BASE_URL and " + "LINEAGEWEAVE_TEST_ORCHESTRATOR_API_KEY to run against a live " + "contextual-orchestrator instance" + ), +) +def test_contextual_orchestrator_extracts_two_sided_keymen_from_ambiguous_prose() -> None: + """A real orchestrator call against a genuinely LLM-ambiguous synthetic + post: one dual-hatted counterparty, two of our people, and an + organization that sent nobody. The assertion is on structure (both + sides present, N affiliations on the counterparty, no invented + Westfield person) rather than exact string spelling. + """ + title, body = ambiguous_keyman_post() + client = ContextualOrchestratorKeymanExtractionClient( + base_url=_ORCHESTRATOR_BASE_URL, api_key=_ORCHESTRATOR_API_KEY + ) + mentions = client.extract(title, body) + + names = " ".join(mention.person_name.lower() for mention in mentions) + assert "priya" in names + assert "jordan" in names or "sam" in names + assert "westfield" not in names + + sides = {mention.person_side_code for mention in mentions} + assert OUR_SIDE in sides + assert COUNTERPARTY in sides + + priya = next(mention for mention in mentions if "priya" in mention.person_name.lower()) + assert priya.person_side_code == COUNTERPARTY + affiliation_blob = " ".join(priya.affiliated_organization_names).lower() + assert "northridge" in affiliation_blob + assert len(priya.affiliated_organization_names) >= 2 diff --git a/uv.lock b/uv.lock index 579b00451..ee688e932 100644 --- a/uv.lock +++ b/uv.lock @@ -438,7 +438,7 @@ wheels = [ [[package]] name = "lineageweave" -version = "0.8.0" +version = "0.9.0" source = { virtual = "." } dependencies = [ { name = "certifi" },