feat: name Allen interval relations on Event Lineage edges (v2.15.0)

ARCHITECTURE.md

@@ -69,6 +69,7 @@ flowchart LR
 | `rankweave_client.py` | Fail-closed RankWeave ranking port (`weighted_reciprocal_rank_fuse` in-process; never invent a fused score or a theta) |
 | `reconstruct.py` | The pipeline: group → candidate window → score → fuse → thread |
 | `lineage_persistence.py` | Flattens reconstruct trees into `post_lineage_edge` row specs (parent, child, fused_score) |
+| `interval_relation.py` | Allen (1983) closed interval relations for those edges. Each post is a point interval on its observed UTC `created_at` day; mutable ticket dates are not Event Lineage evidence. |
 | `knowledge_graph.py` | Random-walk-with-restart relevance + per-node adaptive related-node cutoff (Tong et al., 2006) -- pure graph math, no Postgres |
 | `keyman_extraction.py` | Pluggable LLM extraction of two-sided (our-side/counterparty) person mentions + N:N org affiliations from a post |
 | `entity_relationship_classification.py` | Pluggable LLM classification of a named organization's relationship to the post author (`rel_voc`/`rel_vom`/`rel_vop`/`rel_vocc`/`rel_voco`/`rel_vos`) |
@@ -224,9 +225,12 @@ contextual-orchestrator; persist is `backend/app/keyman_ingestion.py`.
 `GET /api/lineage` returns the ABAC-filtered reconstruct graph
 (`{nodes, edges}`) from persisted `post_lineage_edge` rows. Each node
 includes `group` from the same `reconstruct_group_key()` rebuild uses
-(persisted `thread_group_key`, else process unit, else corp).
+(persisted `thread_group_key`, else process unit, else corp). Each
+direct edge includes `interval_relation_code` / `interval_relation_label`
+(Allen, 1983; ADR 0161) computed from the two posts' observed windows.
 `POST /api/lineage/rebuild` (`post_admin`) re-runs `reconstruct()` over
-every `source_post` and rewrites those edges. Reconstruct grouping is
+every `source_post` and rewrites those edges, then names the interval
+relation in the same transaction. Reconstruct grouping is
 stored on the post as `thread_group_key` / `secondary_grouping_key`
 (not derived from process unit or voc type).
 

CHANGELOG.d/2.15.1-event-lineage-interval-relation.md

@@ -0,0 +1,19 @@
+# 2.15.1 — Name Allen interval relations on Event Lineage edges
+
+## Added
+
+- Persist `interval_relation_code` on `post_lineage_edge` (ADR 0161). After
+  reconstruct chooses a parent, the product names the Allen (1983) relation
+  between the two posts' dated windows. Every post is a point interval on its
+  observed UTC creation day; mutable ticket dates are ignored.
+- After `make seed`, the A-100 pricing follow-up is **Before** the revised
+  quote and delivery question. Click the Before row for the revised quote to
+  open it. The DAG shows those labels as visible text and as a keyboard list.
+  Indirect Keyman links stay unlabeled.
+
+## Changed
+
+- `GET /api/lineage` and `GET /api/posts/{id}/lineage` return the exact
+  lookup label next to the fused score. Opening the child orients the
+  stored parent→child code (Contains → During) so the opened post is the
+  subject of the relation.

CHANGELOG.md

@@ -131,6 +131,15 @@ All notable changes to this project are documented here. Format follows
   on the cited evidence. Open that cited post to read which clock
   matched. Never invent a theta or an event date.
 
+## [2.15.1] - 2026-08-25
+
+### Added
+
+- Event Lineage edges now name the Allen (1983) interval relation between
+  posts' observed UTC creation-day points. Mutable ticket due dates do not
+  rewrite Event Lineage evidence; directed relation rows open the other post,
+  while indirect Keyman links remain unlabeled (ADR 0161).
+
 ## [2.15.0] - 2026-08-25
 
 ### Changed

backend/app/lineage_ingestion.py

@@ -19,6 +19,12 @@
 import asyncpg
 
 from backend.app.post_eligibility import SOURCE_POST_ELIGIBILITY_SQL
+from lineageweave.interval_relation import (
+    INTERVAL_RELATION_LABELS,
+    allen_interval_relation,
+    interval_from_post,
+    interval_relation_from_current,
+)
 from lineageweave.lineage_persistence import lineage_edge_specs
 from lineageweave.models import Edge, Record
 
@@ -72,16 +78,46 @@ def records_from_source_posts(rows: list[Mapping[str, Any]]) -> list[Record]:
     return records
 
 
-async def persist_lineage_edges(conn: asyncpg.Connection, edges: list[Edge]) -> None:
+def interval_relation_code_for_edge(
+    parent_row: Mapping[str, Any], child_row: Mapping[str, Any]
+) -> str:
+    """Allen relation of the parent creation-day point toward the child."""
+    return allen_interval_relation(
+        interval_from_post(parent_row["created_at"]),
+        interval_from_post(child_row["created_at"]),
+    )
+
+
+async def persist_lineage_edges(
+    conn: asyncpg.Connection,
+    edges: list[Edge],
+    points_by_post_id: Mapping[str, Mapping[str, Any]],
+) -> None:
     """Replace ``post_lineage_edge`` with ``edges`` (reconstruct is source of truth)."""
+    missing_point_ids = {
+        post_id
+        for edge in edges
+        for post_id in (edge.parent_id, edge.child_id)
+        if post_id not in points_by_post_id
+    }
+    if missing_point_ids:
+        raise ValueError(
+            "missing observed interval point for post ids: "
+            + ", ".join(sorted(missing_point_ids))
+        )
     await conn.execute("delete from post_lineage_edge")
     for edge in edges:
+        relation_code = interval_relation_code_for_edge(
+            points_by_post_id[edge.parent_id], points_by_post_id[edge.child_id]
+        )
         await conn.execute(
-            "insert into post_lineage_edge (parent_post_id, child_post_id, fused_score) "
-            "values ($1::uuid, $2::uuid, $3)",
+            "insert into post_lineage_edge "
+            "(parent_post_id, child_post_id, fused_score, interval_relation_code) "
+            "values ($1::uuid, $2::uuid, $3, $4)",
             edge.parent_id,
             edge.child_id,
             edge.fused_score,
+            relation_code,
         )
 
 
@@ -232,10 +268,21 @@ async def rebuild_lineage(conn: asyncpg.Connection) -> list[Edge]:
     if weights is None:
         raise ChannelWeightsNotEstimated(active_channels)
     edges = lineage_edge_specs(records_from_source_posts(rows), weights=weights)
-    await persist_lineage_edges(conn, edges)
+    await persist_lineage_edges(conn, edges, {str(row["post_id"]): row for row in rows})
     return edges
 
 
+def _interval_payload(row: Mapping[str, Any]) -> dict[str, Any]:
+    code = row.get("interval_relation_code")
+    if not code:
+        return {}
+    label = row.get("interval_relation_label") or INTERVAL_RELATION_LABELS.get(str(code))
+    payload = {"interval_relation_code": str(code)}
+    if label:
+        payload["interval_relation_label"] = str(label)
+    return payload
+
+
 async def visible_lineage_graph(
     conn: asyncpg.Connection,
     can_see_post,
@@ -256,7 +303,8 @@ async def visible_lineage_graph(
     )
     visible_all = [row for row in posts if can_see_post(row)]
     edge_rows = await conn.fetch(
-        "select parent_post_id, child_post_id, fused_score from post_lineage_edge"
+        "select parent_post_id, child_post_id, fused_score, "
+        "interval_relation_code from post_lineage_edge"
     )
 
     if focus_post_id is None:
@@ -320,12 +368,45 @@ async def visible_lineage_graph(
             "source": str(row["parent_post_id"]),
             "target": str(row["child_post_id"]),
             "fused_score": float(row["fused_score"]),
+            **_interval_payload(row),
         }
         for row in visible_edges
     ]
     return {"nodes": nodes, "edges": edges, "truncated": truncated}
 
 
+async def interval_relations_for_post(
+    conn: asyncpg.Connection, post_id: str
+) -> dict[str, dict[str, Any]]:
+    """Allen labels on direct reconstructed neighbors of ``post_id``."""
+    rows = await conn.fetch(
+        "select parent_post_id, child_post_id, interval_relation_code "
+        "from post_lineage_edge "
+        "where parent_post_id = $1::uuid or child_post_id = $1::uuid",
+        post_id,
+    )
+    current = str(post_id)
+    relations: dict[str, dict[str, Any]] = {}
+    for row in rows:
+        parent_id = str(row["parent_post_id"])
+        child_id = str(row["child_post_id"])
+        other_id = child_id if parent_id == current else parent_id
+        current_is_parent = parent_id == current
+        stored = _interval_payload(row)
+        code = stored.get("interval_relation_code")
+        if not code:
+            continue
+        oriented = interval_relation_from_current(str(code), current_is_parent)
+        relations[other_id] = {
+            "interval_relation_code": oriented,
+            "interval_relation_label": INTERVAL_RELATION_LABELS.get(
+                oriented, stored.get("interval_relation_label")
+            ),
+            "interval_is_parent": current_is_parent,
+        }
+    return relations
+
+
 async def lineage_graphs_for_posts(
     conn: asyncpg.Connection,
     can_see_post,

backend/app/main.py

@@ -171,6 +171,7 @@
 )
 from backend.app.lineage_ingestion import (
     ChannelWeightsNotEstimated,
+    interval_relations_for_post,
     rebuild_lineage,
     visible_lineage_graph,
 )
@@ -2297,22 +2298,27 @@ async def read_post_lineage(
                 list(candidate_ids),
             )
             rows = {str(row["post_id"]): row for row in fetched}
-
-    def _visible_summaries(ids: frozenset[str]) -> list[dict[str, Any]]:
-        return [
-            {
+        direct_intervals = await interval_relations_for_post(conn, post_id)
+
+    def _visible_summaries(ids: frozenset[str], with_intervals: bool = False) -> list[dict[str, Any]]:
+        summaries = []
+        for post_id_ in ids:
+            if post_id_ not in rows or not _can_see_post(account, rows[post_id_]):
+                continue
+            summary = {
                 "post_id": post_id_,
                 "post_title": rows[post_id_]["post_title"],
                 "post_body_excerpt": rows[post_id_].get("post_body_excerpt"),
                 "post_body_truncated": rows[post_id_].get("post_body_truncated", False),
             }
-            for post_id_ in ids
-            if post_id_ in rows and _can_see_post(account, rows[post_id_])
-        ]
+            if with_intervals:
+                summary.update(direct_intervals.get(post_id_, {}))
+            summaries.append(summary)
+        return summaries
 
     return {
         "post_id": post_id,
-        "direct": _visible_summaries(linked.direct),
+        "direct": _visible_summaries(linked.direct, with_intervals=True),
         "indirect": _visible_summaries(linked.indirect),
     }
 

backend/tests/test_api.py

@@ -138,6 +138,11 @@
     / "migrations"
     / "0135_lineage_channel_weight.sql"
 )
+_INTERVAL_RELATION_MIGRATION = (
+    Path(__file__).resolve().parents[2]
+    / "migrations"
+    / "0140_post_lineage_interval_relation.sql"
+)
 _CHANNEL_WEIGHT_UNION_MIGRATION = (
     Path(__file__).resolve().parents[2]
     / "migrations"
@@ -306,6 +311,7 @@ def seeded_db(demo_analyst_token):
             cur.execute(_TOPIC_LINEAGE_RESULT_MIGRATION.read_text())
             cur.execute(_TOPIC_LINEAGE_VALIDATE_MIGRATION.read_text())
             cur.execute(_CHANNEL_WEIGHT_MIGRATION.read_text())
+            cur.execute(_INTERVAL_RELATION_MIGRATION.read_text())
             cur.execute(_CHANNEL_WEIGHT_UNION_MIGRATION.read_text())
             cur.execute(_PAIR_JUDGMENT_MIGRATION.read_text())
             # Product reconstruction fails closed without an ACTIVATED
@@ -4311,6 +4317,94 @@ def test_rebuild_lineage_recovers_the_a100_fork(client, demo_analyst_token, seed
     assert "Delivery schedule question raised" in direct_titles
 
 
+def test_rebuild_lineage_ignores_mutable_ticket_dates(
+    client, demo_analyst_token, seeded_db
+) -> None:
+    """Manual ticket dates do not replace observed post chronology."""
+    from scripts.seed_demo_data import (
+        _seed_fixture_tickets,
+        insert_fixture_source_posts,
+    )
+
+    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'), "
+                "('voc_type', 'vom', 'Voice of Market') "
+                "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 process_unit (corporate_entity_id, process_unit_code, process_unit_name) "
+                "select corporate_entity_id, 'TEST-PU-INTERVAL', 'Interval thread' "
+                "from source_post where post_id = %s returning process_unit_id",
+                (seeded_db["own_private_post_id"],),
+            )
+            process_unit_id = cur.fetchone()[0]
+            cur.execute(
+                "select author_account_id, corporate_entity_id from source_post where post_id = %s",
+                (seeded_db["own_private_post_id"],),
+            )
+            author_id, corp_id = cur.fetchone()
+            insert_fixture_source_posts(cur, author_id, corp_id, process_unit_id)
+            _seed_fixture_tickets(cur)
+    finally:
+        admin_conn.close()
+
+    rebuild = client.post("/api/lineage/rebuild", headers={"Authorization": f"Bearer {demo_analyst_token}"})
+    assert rebuild.status_code == 200, rebuild.text
+
+    graph = client.get("/api/lineage", headers={"Authorization": f"Bearer {demo_analyst_token}"})
+    assert graph.status_code == 200
+    body = graph.json()
+    nodes = {node["label"]: node for node in body["nodes"]}
+    fork = nodes["Pricing renegotiation follow-up"]
+    quote = nodes["Pricing renegotiation: revised quote sent"]
+    delivery = nodes["Delivery schedule question raised"]
+    quote_edge = next(
+        edge
+        for edge in body["edges"]
+        if edge["source"] == fork["id"] and edge["target"] == quote["id"]
+    )
+    delivery_edge = next(
+        edge
+        for edge in body["edges"]
+        if edge["source"] == fork["id"] and edge["target"] == delivery["id"]
+    )
+    assert quote_edge["interval_relation_code"] == "interval_before"
+    assert quote_edge["interval_relation_label"] == "Before"
+    assert delivery_edge["interval_relation_code"] == "interval_before"
+    assert delivery_edge["interval_relation_label"] == "Before"
+
+    per_post = client.get(
+        f"/api/posts/{fork['id']}/lineage",
+        headers={"Authorization": f"Bearer {demo_analyst_token}"},
+    )
+    assert per_post.status_code == 200
+    by_title = {post["post_title"]: post for post in per_post.json()["direct"]}
+    assert by_title["Pricing renegotiation: revised quote sent"]["interval_relation_code"] == "interval_before"
+    assert by_title["Delivery schedule question raised"]["interval_relation_code"] == "interval_before"
+
+    from_quote = client.get(
+        f"/api/posts/{quote['id']}/lineage",
+        headers={"Authorization": f"Bearer {demo_analyst_token}"},
+    )
+    assert from_quote.status_code == 200
+    quote_direct = {post["post_id"]: post for post in from_quote.json()["direct"]}
+    assert quote_direct[fork["id"]]["interval_relation_code"] == "interval_after"
+    assert quote_direct[fork["id"]]["interval_relation_label"] == "After"
+    assert quote_direct[fork["id"]]["interval_is_parent"] is False
+
+
 def test_lineage_graph_hides_other_corp_private_posts(client, demo_analyst_token, seeded_db) -> None:
     response = client.get("/api/lineage", headers={"Authorization": f"Bearer {demo_analyst_token}"})
     assert response.status_code == 200
@@ -4510,6 +4604,22 @@ def test_post_activity_is_empty_before_any_mutation(client, demo_analyst_token,
     assert response.json()["events"] == []
 
 
+def test_post_activity_requires_post_read(client, demo_analyst_token, seeded_db) -> None:
+    admin_conn = psycopg2.connect(seeded_db["dsn"])
+    admin_conn.autocommit = True
+    try:
+        with admin_conn.cursor() as cur:
+            cur.execute("delete from role_permission where permission_code = 'post_read'")
+    finally:
+        admin_conn.close()
+
+    response = client.get(
+        f"/api/posts/{seeded_db['own_private_post_id']}/activity",
+        headers={"Authorization": f"Bearer {demo_analyst_token}"},
+    )
+    assert response.status_code == 403
+
+
 def test_ticket_mutations_publish_real_events_to_the_activity_feed(
     client, demo_analyst_token, seeded_db
 ) -> None: