feat(orchestrator): deliver endpoint routing and bounded admission runtime

CHANGELOG.md

@@ -6,7 +6,32 @@ All notable changes to this project are documented here. Format follows
 
 ## [Unreleased]
 
-### Added
+- Post-content recovery now keys every initial attempt, retry, and stale lease
+  by its exact eligibility instant, so work that becomes due after the durable
+  cursor advances is reached without waiting for a full ledger wrap.
+
+- Global Ask public verification now admits only bounded, provenance-bearing
+  persisted claims attached to exact cited public posts; missing admission
+  fails closed without token-overlap egress.
+
+### Added
+
+- Temporal topic influence now has a durable external-production path: the
+  worker binds the exact completed TEPP artifact, posterior draws, and
+  business-unit/PU/team/person memberships into a content-addressed request,
+  then persists only a complete, converged, identified, parity-passed
+  fast-mlsirm result. Missing owner transport, partial rows, or digest mismatch
+  remains unavailable without local scoring. Time-valid membership slices
+  remain distinct; incomplete evidence enters an event-woken awaiting state;
+  expired work is reclaimed only from its declared request/lease contract,
+  whose lease must strictly exceed the request timeout for persistence; and
+  every terminal transition matches a unique lease token. Evidence changed
+  during computation releases a fresh request automatically.
+  LineageWeave-owned request and membership bytes and producer-owned result
+  bytes are SHA-256 verified before parsing, so admission never depends on
+  cross-language JSON reserialization. Other
+  retries use only an exact remote delay or an explicit operator requeue
+  (ADR 0210).
 
 - Evidence Operations now presents cited claim, rebid, handover, external,
   product, and Voice evidence with explicit unavailable states and source-open
@@ -224,6 +249,13 @@ All notable changes to this project are documented here. Format follows
 
 ### Fixed
 
+- Public-claim provenance validation now selects its sole UUID binding without
+  calling PostgreSQL's unavailable `min(uuid)` aggregate, so fresh schema
+  replays accept valid provenance-bound public claims.
+- Post-content wake-up recovery now advances through every ready ledger page
+  with a deterministic keyset, while Valkey trims only entries already
+  consumed by the worker. Large backfills can no longer replay the same first
+  page until a later queued record starves or its unread wake-up is trimmed.
 - Full-corpus Event Lineage rebuilds now count candidate pairs before provider
   work and omit the optional LLM channel above the 5,000-pair ADR budget,
   preventing millions of synchronous orchestrator calls while retaining one

backend/app/config.py

@@ -62,6 +62,11 @@ class Settings:
     source_research_maximum_results: int | None
     tepp_transport_url: str
     tepp_api_key: str
+    topic_influence_transport_url: str
+    topic_influence_api_key: str
+    topic_influence_request_timeout_seconds: int | None
+    topic_influence_lease_timeout_seconds: int | None
+    topic_influence_poll_seconds: int | None
     caldav_base_url: str
     naruon_calendar_base_url: str
     naruon_calendar_service_token: str
@@ -215,6 +220,21 @@ def load_settings() -> Settings:
         ),
         tepp_transport_url=os.environ.get("TEPP_TRANSPORT_URL", ""),
         tepp_api_key=os.environ.get("TEPP_API_KEY", "").strip(),
+        topic_influence_transport_url=os.environ.get(
+            "TOPIC_INFLUENCE_TRANSPORT_URL", ""
+        ).strip(),
+        topic_influence_api_key=os.environ.get(
+            "TOPIC_INFLUENCE_API_KEY", ""
+        ).strip(),
+        topic_influence_request_timeout_seconds=_optional_positive_int(
+            "TOPIC_INFLUENCE_REQUEST_TIMEOUT_SECONDS"
+        ),
+        topic_influence_lease_timeout_seconds=_optional_positive_int(
+            "TOPIC_INFLUENCE_LEASE_TIMEOUT_SECONDS"
+        ),
+        topic_influence_poll_seconds=_optional_positive_int(
+            "TOPIC_INFLUENCE_POLL_SECONDS"
+        ),
         caldav_base_url=os.environ.get("CALDAV_BASE_URL", "").strip(),
         naruon_calendar_base_url=os.environ.get("NARUON_CALENDAR_BASE_URL", "").strip(),
         naruon_calendar_service_token=os.environ.get(

backend/app/global_ask_queue.py

@@ -38,7 +38,6 @@
     ClaimVerificationClient,
     ClaimVerificationResult,
     NullClaimVerificationClient,
-    public_claim_candidates,
 )
 from lineageweave.embedding_client import EmbeddingClient, NullEmbeddingClient
 from lineageweave.http_client import HttpClientError
@@ -52,6 +51,10 @@
     cited_post_summaries,
     historical_body_limitations,
 )
+from lineageweave.public_claim_envelope import (
+    PersistedPublicClaimEnvelope,
+    envelope_from_authorized_row,
+)
 from lineageweave.semantic_query import NullSemanticQueryClient, SemanticQueryClient
 from lineageweave.temporal_expressions import resolve_korean_relative_time
 
@@ -103,6 +106,33 @@
 
 _logger = logging.getLogger(__name__)
 
+_AUTHORIZED_PUBLIC_CLAIM_ENVELOPES_SQL = """
+    select envelope.public_claim_envelope_id,
+           envelope.source_post_id,
+           envelope.claim_kind_code,
+           envelope.claim_text
+      from public_claim_envelope envelope
+      join source_post post on post.post_id = envelope.source_post_id
+      join provenance_assertion assertion
+        on assertion.assertion_id = envelope.provenance_assertion_id
+       and assertion.relation_code = 'prov_was_derived_from'
+     where envelope.egress_eligible
+       and post.visibility_code = 'public'
+       and envelope.source_post_id = any($1::uuid[])
+       and exists (
+           select 1
+             from provenance_resource_binding evidence
+            where evidence.resource_id = assertion.object_resource_id
+              and evidence.node_type_code = 'node_post'
+              and evidence.node_id = envelope.source_post_id
+       )
+       and ($2::timestamptz is null or (
+            envelope.created_at <= $2 and post.created_at <= $2
+       ))
+     order by envelope.created_at, envelope.public_claim_envelope_id
+     limit 4
+"""
+
 
 class _SafeJobError(Exception):
     """Failure whose bounded message is safe to persist for the requester."""
@@ -188,16 +218,20 @@ async def _verify_public_claims(
     *,
     verify_external: bool,
     client: ClaimVerificationClient,
+    persisted_envelopes: tuple[PersistedPublicClaimEnvelope, ...] = (),
 ) -> tuple[str, tuple[ClaimVerificationResult, ...]]:
-    """Verify only cited claims explicitly marked safe for public egress."""
+    """Verify only cited claims explicitly marked safe for public egress.
+
+    Only persisted admission envelopes may cross the public verifier. Omitting
+    them fails closed; question-token overlap is not an admission mechanism.
+    """
 
     if not verify_external:
         return VERIFICATION_SKIPPED, ()
     cited_ids = frozenset(cited_post_ids)
+    claims = tuple(envelope.verification_candidate() for envelope in persisted_envelopes)
     claims = tuple(
-        claim
-        for claim in public_claim_candidates(sources, question)
-        if set(claim.source_post_ids).issubset(cited_ids)
+        claim for claim in claims if set(claim.source_post_ids).issubset(cited_ids)
     )
     if not claims:
         return VERIFICATION_NO_PUBLIC_CLAIMS, ()
@@ -219,6 +253,28 @@ async def _verify_public_claims(
     )
 
 
+async def load_authorized_public_claim_envelopes(
+    conn: asyncpg.Connection,
+    cited_post_ids: list[str],
+    *,
+    knowledge_cutoff: datetime | None,
+) -> tuple[PersistedPublicClaimEnvelope, ...]:
+    """Load bounded persisted claims for exact cited public evidence posts."""
+
+    if not cited_post_ids:
+        return ()
+    rows = await conn.fetch(
+        _AUTHORIZED_PUBLIC_CLAIM_ENVELOPES_SQL,
+        cited_post_ids,
+        knowledge_cutoff,
+    )
+    return tuple(
+        envelope
+        for row in rows
+        if (envelope := envelope_from_authorized_row(row)) is not None
+    )
+
+
 async def load_job_visibility(
     conn: asyncpg.Connection, job_id: str, account_id: str
 ) -> tuple[set[str], set[str], bool, bool]:
@@ -372,6 +428,7 @@ def can_see(row: asyncpg.Record) -> bool:
             [],
             verify_external=verify_external,
             client=verification_client,
+            persisted_envelopes=(),
         )
         delivery = build_ask_delivery("", (), ())
         return {
@@ -433,14 +490,16 @@ def can_see(row: asyncpg.Record) -> bool:
             _ASK_RETRY_MESSAGE,
         ) from exc
     cited_ids = list(answer.cited_post_ids)
-    verification_status, external_claims = await _verify_public_claims(
-        question_text,
-        usable_sources,
-        cited_ids,
-        verify_external=verify_external,
-        client=verification_client,
-    )
     async with pool.acquire() as conn:
+        persisted_envelopes = (
+            await load_authorized_public_claim_envelopes(
+                conn,
+                cited_ids,
+                knowledge_cutoff=knowledge_cutoff,
+            )
+            if verify_external
+            else ()
+        )
         if knowledge_cutoff is None:
             lineage_graph = await lineage_graphs_for_posts(conn, can_see, cited_ids)
             images = await cited_post_images(conn, cited_ids)
@@ -452,6 +511,14 @@ def can_see(row: asyncpg.Record) -> bool:
             cited_ids,
             checked_by=knowledge_cutoff,
         )
+    verification_status, external_claims = await _verify_public_claims(
+        question_text,
+        usable_sources,
+        cited_ids,
+        verify_external=verify_external,
+        client=verification_client,
+        persisted_envelopes=persisted_envelopes,
+    )
     cited_posts = cited_post_summaries(usable_sources, cited_ids)
     cited_events = cited_post_events(usable_sources, cited_ids)
     cited_evidence = cited_post_evidence(usable_sources, cited_ids)

backend/app/operations_dashboard.py

@@ -189,14 +189,6 @@ async def fetch_operations_dashboard(
                 )
         )
         select (select count(*) from visible_post) as total_post_count,
-               (select count(*)
-                  from post_summary_event summary_event
-                 where exists (
-                     select 1 from classified
-                      where classified.post_id = summary_event.post_id
-                        and ($5::boolean is false
-                             or classified.case_kind_code = 'external_information')
-                 )) as total_event_count,
                (select count(distinct post_id) from classified
                  where case_kind_code = 'external_information') as external_post_count,
                (select count(*) from scoped_post
@@ -225,10 +217,7 @@ async def fetch_operations_dashboard(
                coalesce(post.event_occurred_at, post.created_at) as occurred_at,
                coalesce(nullif(btrim(post.source_project_name), ''), project.primary_project_name)
                    as project_name,
-               coalesce(project.project_names, array[]::text[]) as project_names,
-               (select count(*)::int
-                  from post_summary_event summary_event
-                 where summary_event.post_id = classification.post_id) as event_count
+               coalesce(project.project_names, array[]::text[]) as project_names
           from operations_case_classification classification
           join source_post post on post.post_id = classification.post_id
           join source_post evidence_post
@@ -439,10 +428,17 @@ async def fetch_operations_dashboard(
     external = int(metrics["external_post_count"])
     case_post_ids: dict[str, set[str]] = {}
     case_event_counts: dict[str, int] = {}
+    counted_case_keys: set[tuple[str, str]] = set()
     for row in case_rows:
         kind = row["case_kind_code"]
-        case_post_ids.setdefault(kind, set()).add(str(row["post_id"]))
-        case_event_counts[kind] = case_event_counts.get(kind, 0) + int(row["event_count"])
+        post_id = str(row["post_id"])
+        case_post_ids.setdefault(kind, set()).add(post_id)
+        key = (post_id, kind)
+        if key not in counted_case_keys:
+            case_event_counts[kind] = case_event_counts.get(kind, 0) + len(
+                milestones.get(key, ())
+            )
+            counted_case_keys.add(key)
     projected_cases = []
     lifecycle_metrics = {
         lifecycle_code: {
@@ -493,7 +489,7 @@ async def fetch_operations_dashboard(
         "period_end": period_end.isoformat() if period_end else None,
         "period_time_axis_code": "event_occurred_at",
         "total_post_count": total,
-        "total_event_count": int(metrics["total_event_count"]),
+        "total_event_count": sum(case_event_counts.values()),
         "external_post_count": external,
         "external_percent": external * 100 / total if total else 0.0,
         "pending_analysis_count": int(metrics["pending_analysis_count"]),