feat(warpline): lifecycle facts on entity-assoc reverse-lookup + atomic reverify ingest

CHANGELOG.md

@@ -5,6 +5,34 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Added
+
+- **Warpline seam — issue lifecycle facts on the entity-association
+  reverse-lookup.** `GET /api/entity-associations?entity_id=<sei>` (and the MCP
+  and data-layer reverse lookups) now enriches each binding row with the bound
+  issue's `claimed_at`, `closed_at`, `status`, and `status_category`, so warpline
+  can correlate "changed since the issue was claimed/closed" against its own
+  changed-set in a single round trip. `closed_at` is the proven-good signal
+  ("issue closed at commit X"); Filigree exposes the resolution timestamp and
+  stores no commit SHA (warpline maps timestamp→commit on its side). Implemented
+  as a separate enriched projection over a `LEFT JOIN` to `issues` — the shared
+  mapper, the forward per-issue list, the add-response, and the governance
+  closure gate are untouched and byte-identical; an orphaned binding still
+  returns (null facts). Additive and Loomweave-safe (the consumer ignores
+  unknown fields); the entity-associations contract fixture is bumped to v2.
+
+### Fixed
+
+- **Warpline reverify ingest filed an issue and bound its SEI
+  non-atomically.** `warpline_worklist_ingest` created the issue and attached
+  its SEI association in two separate transactions, so a non-retryable storage
+  error on the bind left the issue FILED-but-UNBOUND — and the next ingest then
+  re-filed a duplicate because the loop-closure contract keys on the SEI
+  binding. File + bind now commit together in one transaction via
+  `create_issue`'s inline ADR-029 bind path.
+
 ## [3.0.1] - 2026-06-18
 
 ### Fixed

src/filigree/db_entity_associations.py

@@ -23,14 +23,15 @@
 from __future__ import annotations
 
 from collections.abc import Mapping
-from typing import Any, TypedDict
+from typing import Any, TypedDict, cast
 
 from filigree.db_base import DBMixinProtocol, _in_immediate_tx, _now_iso, _retry_busy
 from filigree.types.core import (
     ContentHash,
     ISOTimestamp,
     IssueId,
     LoomweaveEntityId,
+    StatusCategory,
     make_content_hash,
     make_issue_id,
     make_loomweave_entity_id,
@@ -71,6 +72,30 @@ class EntityAssociationRow(TypedDict):
     signed_content_hash: ContentHash | None
 
 
+class EntityAssociationByEntityRow(EntityAssociationRow):
+    """A reverse-lookup row (``list_associations_by_entity``) enriched with the
+    bound issue's lifecycle facts.
+
+    The warpline<->filigree seam (B contract): warpline reads the reverse lookup
+    to learn which issues are bound to a code entity, then correlates "changed
+    since the issue was claimed/closed" against its own changed-set. Carrying the
+    lifecycle anchors here lets it do that in one round trip. ``closed_at`` is the
+    proven-good signal ("issue closed at commit X"); Filigree exposes the
+    resolution timestamp verbatim and stores no commit SHA — warpline maps the
+    timestamp to a commit on its own side.
+
+    All four fields are ``None`` for an *orphaned* binding (the issue row is
+    absent — a LEFT JOIN, not INNER, so the binding is still returned rather than
+    dropped). These keys appear ONLY on the reverse projection; the forward
+    per-issue list (``list_entity_associations``) stays a pure binding row.
+    """
+
+    claimed_at: ISOTimestamp | None
+    closed_at: ISOTimestamp | None
+    status: str | None
+    status_category: StatusCategory | None
+
+
 class GovernedAssociationRemovalError(ValueError):
     """Refused: a caller tried to delete a Legis-signed (governed) binding.
 
@@ -390,12 +415,38 @@ def list_entity_associations(self, issue_id: IssueId) -> list[EntityAssociationR
         ).fetchall()
         return [_row_to_entity_association(r) for r in rows]
 
+    def _row_to_by_entity_association(
+        self, r: Mapping[str, Any], *, current_content_hash: str | None = None
+    ) -> EntityAssociationByEntityRow:
+        """Enrich a reverse-lookup row with the bound issue's lifecycle facts.
+
+        The base binding projection is byte-identical to the forward list (the
+        shared :func:`_row_to_entity_association` is untouched); only the reverse
+        surface carries the LEFT-JOINed ``issues`` columns. ``status`` /
+        ``closed_at`` / ``claimed_at`` are ``None`` for an orphaned binding (issue
+        row absent), and ``status_category`` is then ``None`` too, so the row is
+        still returned rather than dropped.
+        """
+        base = _row_to_entity_association(r, current_content_hash=current_content_hash)
+        status = r["issue_status"]
+        issue_type = r["issue_type"]
+        claimed_at = r["issue_claimed_at"]
+        closed_at = r["issue_closed_at"]
+        enriched = cast("EntityAssociationByEntityRow", dict(base))
+        enriched["claimed_at"] = ISOTimestamp(claimed_at) if claimed_at else None
+        enriched["closed_at"] = ISOTimestamp(closed_at) if closed_at else None
+        enriched["status"] = status
+        enriched["status_category"] = (
+            self._resolve_status_category(issue_type, status) if status is not None and issue_type is not None else None
+        )
+        return enriched
+
     def list_associations_by_entity(
         self,
         entity_id: LoomweaveEntityId,
         *,
         current_content_hash: ContentHash | str | None = None,
-    ) -> list[EntityAssociationRow]:
+    ) -> list[EntityAssociationByEntityRow]:
         """Return all issue bindings for a given Loomweave entity.
 
         The reverse of :meth:`list_entity_associations`: given an
@@ -414,16 +465,24 @@ def list_associations_by_entity(
         entity_id = make_loomweave_entity_id(entity_id)
         if current_content_hash is not None:
             current_content_hash = make_content_hash(current_content_hash)
+        # LEFT JOIN (not INNER) so an orphaned binding — issue row gone, a state
+        # warpline_consumer explicitly contemplates ("binding outlived its
+        # issue") — is still returned with null lifecycle facts rather than
+        # silently dropped. The issue columns are aliased to avoid any collision
+        # and to read explicitly in the enrichment mapper.
         rows = self.conn.execute(
             """
-            SELECT issue_id, loomweave_entity_id, entity_kind, content_hash_at_attach,
-                   attached_at, attached_by, migration_orphaned_at, signature, signoff_seq,
-                   signed_content_hash
-            FROM entity_associations
-            WHERE loomweave_entity_id = ?
-            ORDER BY attached_at ASC, issue_id ASC
+            SELECT ea.issue_id, ea.loomweave_entity_id, ea.entity_kind, ea.content_hash_at_attach,
+                   ea.attached_at, ea.attached_by, ea.migration_orphaned_at, ea.signature,
+                   ea.signoff_seq, ea.signed_content_hash,
+                   i.claimed_at AS issue_claimed_at, i.closed_at AS issue_closed_at,
+                   i.status AS issue_status, i.type AS issue_type
+            FROM entity_associations ea
+            LEFT JOIN issues i ON i.id = ea.issue_id
+            WHERE ea.loomweave_entity_id = ?
+            ORDER BY ea.attached_at ASC, ea.issue_id ASC
             """,
             (entity_id,),
         ).fetchall()
         current_hash = str(current_content_hash) if current_content_hash is not None else None
-        return [_row_to_entity_association(r, current_content_hash=current_hash) for r in rows]
+        return [self._row_to_by_entity_association(r, current_content_hash=current_hash) for r in rows]

src/filigree/warpline_consumer.py

@@ -35,7 +35,7 @@
 from collections.abc import Mapping
 from typing import TYPE_CHECKING, Any
 
-from filigree.types.core import make_content_hash, make_issue_id, make_loomweave_entity_id
+from filigree.types.core import make_loomweave_entity_id
 
 if TYPE_CHECKING:
     from filigree.core import FiligreeDB
@@ -203,19 +203,24 @@ def ingest_reverify_worklist(
             result["prior_closed_issue_ids"] = closed_issue_ids
 
         if apply:
+            # File the issue and bind its SEI in a SINGLE transaction via
+            # create_issue's inline ADR-029 bind (SEAM SEI-on-create). A separate
+            # add_entity_association call would commit the issue first, so a
+            # storage failure on the bind would leave the issue FILED-but-UNBOUND
+            # — and the next ingest would re-file a duplicate, because the
+            # loop-closure contract keys on the SEI binding. The inline path makes
+            # file+bind atomic, and (re)applies the same UNVERIFIED sentinel for a
+            # blank hash. content_hash here is already resolved (a real hash or the
+            # warpline sentinel), so it is preserved verbatim.
             issue = db.create_issue(
                 f"Reverify: {locator if locator else sei}",
                 type="task",
                 priority=priority,
                 description=_build_description(raw, sei, locator),
                 labels=list(PRODUCER_LABELS),
                 actor=actor,
-            )
-            db.add_entity_association(
-                make_issue_id(issue.id),
-                entity_id,
-                make_content_hash(content_hash),
-                actor=actor,
+                entity_id=str(entity_id),
+                content_hash=content_hash,
                 entity_kind=ENTITY_KIND,
             )
             result["issue_id"] = issue.id

tests/core/test_entity_associations.py

@@ -363,6 +363,69 @@ def test_attach_persists_signature_and_signoff_seq(self, db: FiligreeDB) -> None
         assert by_entity[0]["signature"] == "deadbeef"
         assert by_entity[0]["signoff_seq"] == 7
 
+
+class TestReverseLookupLifecycleFacts:
+    """b-ii (warpline seam): the reverse lookup enriches each binding with the
+    bound issue's lifecycle facts — ``claimed_at``, ``closed_at``, ``status``,
+    and ``status_category`` — so warpline can correlate "changed since
+    claimed/closed" against its own changed-set in one round trip. closed_at is
+    the proven-good signal (issue closed at commit X); warpline maps the
+    timestamp to a commit on its side.
+
+    The forward per-issue list (``list_entity_associations``) stays a pure
+    binding projection and must NOT grow these keys — the shared mapper is
+    untouched; only the reverse query joins ``issues``.
+    """
+
+    def test_open_issue_row_exposes_null_close_and_open_category(self, db: FiligreeDB) -> None:
+        issue = db.create_issue("open work", priority=2)
+        db.add_entity_association(issue.id, "py:func:open-target", content_hash="h1", actor="alice")
+        (row,) = db.list_associations_by_entity("py:func:open-target")
+        assert row["closed_at"] is None
+        assert row["claimed_at"] is None
+        assert row["status"] == issue.status
+        assert row["status_category"] == "open"
+
+    def test_closed_issue_row_exposes_closed_at_and_done_category(self, db: FiligreeDB) -> None:
+        issue = db.create_issue("to close", priority=2)
+        db.add_entity_association(issue.id, "py:func:closed-target", content_hash="h1", actor="alice")
+        db.close_issue(issue.id, reason="done")
+        (row,) = db.list_associations_by_entity("py:func:closed-target")
+        assert row["closed_at"] is not None
+        assert row["status_category"] == "done"
+
+    def test_claimed_issue_row_exposes_claimed_at(self, db: FiligreeDB) -> None:
+        issue = db.create_issue("to claim", priority=2)
+        db.add_entity_association(issue.id, "py:func:claimed-target", content_hash="h1", actor="alice")
+        db.claim_issue(issue.id, assignee="alice")
+        (row,) = db.list_associations_by_entity("py:func:claimed-target")
+        assert row["claimed_at"] is not None
+
+    def test_orphaned_binding_still_returns_with_null_facts(self, db: FiligreeDB) -> None:
+        """LEFT JOIN, not INNER: a binding whose issue row is absent must still
+        appear (warpline_consumer contemplates 'binding outlived its issue'),
+        with null lifecycle facts rather than being dropped."""
+        issue = db.create_issue("soon orphaned", priority=2)
+        db.add_entity_association(issue.id, "py:func:orphan-target", content_hash="h1", actor="alice")
+        # Delete the issue row directly, leaving the binding (bypassing cascade).
+        db.conn.execute("PRAGMA foreign_keys = OFF")
+        db.conn.execute("DELETE FROM issues WHERE id = ?", (issue.id,))
+        db.conn.execute("PRAGMA foreign_keys = ON")
+        (row,) = db.list_associations_by_entity("py:func:orphan-target")
+        assert row["issue_id"] == issue.id
+        assert row["closed_at"] is None
+        assert row["status"] is None
+        assert row["status_category"] is None
+
+    def test_forward_list_omits_lifecycle_facts(self, db: FiligreeDB) -> None:
+        issue = db.create_issue("forward", priority=2)
+        db.add_entity_association(issue.id, "py:func:fwd-target", content_hash="h1", actor="alice")
+        (fwd,) = db.list_entity_associations(issue.id)
+        assert "closed_at" not in fwd
+        assert "claimed_at" not in fwd
+        assert "status" not in fwd
+        assert "status_category" not in fwd
+
     def test_attach_without_signature_stores_null(self, db: FiligreeDB) -> None:
         issue = db.create_issue("Ungoverned work", priority=2)
         row = db.add_entity_association(issue.id, "sei:plain", content_hash="h1", actor="x")

tests/fixtures/contracts/entity-associations-response.json

@@ -2,7 +2,7 @@
   "_meta": {
     "contract": "filigree-entity-associations-response",
     "endpoint": "GET /api/entity-associations?entity_id={entity_id}",
-    "fixture_version": 1,
+    "fixture_version": 2,
     "stability": "normative",
     "authority": "Weft G15 / ADR-029 EntityAssociation producer-consumer conformance",
     "producer": "Filigree dashboard_routes.entities.api_list_associations_by_entity",
@@ -12,8 +12,8 @@
       "loomweave/docs/federation/fixtures/filigree-entity-associations-response.json"
     ],
     "verification": "Filigree: uv run pytest tests/api/test_entity_associations.py -q; Loomweave: cargo test -p loomweave-federation filigree::tests::parses_canonical_filigree_entity_association_fixture",
-    "updated": "2026-06-13",
-    "description": "Canonical live Filigree reverse-lookup response for EntityAssociation rows. The producer test emits this body through Filigree's HTTP route; the consumer test deserializes the same body in Loomweave."
+    "updated": "2026-06-24",
+    "description": "Canonical live Filigree reverse-lookup response for EntityAssociation rows. The producer test emits this body through Filigree's HTTP route; the consumer test deserializes the same body in Loomweave. v2 (warpline seam): each row now also carries the bound issue's lifecycle facts (claimed_at, closed_at, status, status_category) for warpline's 'changed since claimed/closed' correlation; all four are null for an orphaned binding. Additive — Loomweave's consumer ignores unknown fields (no serde deny_unknown_fields), so a v1 consumer still parses v2. The loomweave repo-local copy should be synced for documentation parity (non-breaking)."
   },
   "shape_decl": {
     "kind": "filigree-entity-associations-fixture-shapes",
@@ -54,7 +54,11 @@
               "freshness_status": "unknown",
               "signature": null,
               "signoff_seq": null,
-              "signed_content_hash": null
+              "signed_content_hash": null,
+              "claimed_at": null,
+              "closed_at": null,
+              "status": "open",
+              "status_category": "open"
             }
           ]
         }

tests/test_warpline_consumer.py

@@ -10,6 +10,8 @@
 
 from typing import Any
 
+import pytest
+
 from filigree.core import FiligreeDB
 from filigree.warpline_consumer import (
     ENTITY_KIND,
@@ -158,3 +160,27 @@ def test_mixed_worklist_summary(self, db: FiligreeDB) -> None:
             apply=True,
         )
         assert report["summary"] == {"filed": 1, "linked": 1, "skipped": 1, "total": 3}
+
+
+class TestFileAndBindAtomicity:
+    """A filed item must create the issue and bind its SEI in a SINGLE
+    transaction. If the bind fails, the whole item rolls back — no issue may be
+    left FILED-but-UNBOUND, otherwise the next ingest sees the SEI as untracked
+    and re-files a duplicate, breaking the loop-closure contract.
+    """
+
+    def test_bind_failure_rolls_back_the_filed_issue(self, db: FiligreeDB, monkeypatch: pytest.MonkeyPatch) -> None:
+        before = len(db.list_issues())
+
+        def boom(*args: Any, **kwargs: Any) -> None:
+            raise RuntimeError("simulated non-retryable storage failure on bind")
+
+        monkeypatch.setattr(db, "add_entity_association", boom)
+
+        with pytest.raises(RuntimeError, match="storage failure on bind"):
+            ingest_reverify_worklist(db, _worklist(_item("loomweave:eid:ATOMIC")), apply=True)
+
+        # Atomic: the issue insert rolled back together with the failed bind —
+        # no orphaned FILED-but-UNBOUND task, and nothing bound to the SEI.
+        assert len(db.list_issues()) == before
+        assert db.list_associations_by_entity("loomweave:eid:ATOMIC") == []