refactor(activity): qualify Valkey activity identifiers

backend/app/main.py

@@ -42,6 +42,7 @@
 from backend.app.activity_stream import (
     create_valkey_client,
     get_valkey,
+    index_legacy_activity_stream_aliases,
     publish_activity_event,
     read_activity_events,
     ticket_created_summary,
@@ -297,6 +298,7 @@ async def lifespan(app: FastAPI):
         pool = await create_pool(settings.database_url)
         app.state.pool = pool
         valkey = create_valkey_client(settings.valkey_url)
+        await index_legacy_activity_stream_aliases(valkey)
         app.state.valkey = valkey
         analysis_worker = asyncio.create_task(
             run_analysis_run_worker(

docs/adr/0363-canonical-activity-stream-identity.md

@@ -0,0 +1,52 @@
+# ADR 0363: Canonical activity-stream identity
+
+- Status: Proposed
+- Date: 2026-09-04
+
+## Context
+
+`source_post.post_id` is a PostgreSQL `uuid`, while historical HTTP requests
+could spell the same UUID with upper-case digits, braces, omitted hyphens, or
+other PostgreSQL-accepted hyphen placement. Earlier activity writes embedded
+the request spelling in a Valkey stream key. Canonicalizing only new writes can
+therefore strand authorized history, and enumerating selected spellings cannot
+cover PostgreSQL's input grammar.
+
+The activity feed is a retained operational projection. It is not a source of
+record truth, and its compatibility work must not broaden a request into a
+keyspace scan.
+
+## Decision
+
+New activity writes use PostgreSQL's canonical lower-case, hyphenated UUID
+form. Before the application begins serving reads, startup scans the existing
+activity-key namespace once and records every actually present, UUID-equivalent
+legacy key in a durable canonical-post alias set. A request reads the canonical
+stream and only the aliases in that set. Non-UUID synthetic and legacy keys
+retain exact spelling.
+
+Cross-stream ordering uses the persisted millisecond component and declared
+canonical-first precedence for equal milliseconds. Stream-local sequence
+numbers never become a global chronology. Alias event identifiers remain
+namespaced so two streams cannot emit the same public identity.
+
+WATCH retries, retained-window read limits, and alias-index startup are fixed
+product contracts. Failure to build the alias index makes the application
+unready; the product does not silently hide historical activity.
+
+## Consequences
+
+- Canonical reads retain every historical UUID spelling that actually exists.
+- Request-time work is limited to the canonical stream and its durable aliases.
+- Startup performs a cursor scan and must finish before readiness.
+- The alias index is additional derived Valkey state and can be rebuilt from
+  retained activity keys.
+
+## Alternatives considered
+
+- Enumerate common UUID spellings: rejected because PostgreSQL accepts more
+  forms than a finite hand-picked list would honestly cover.
+- Scan the activity keyspace on each request: rejected because latency and work
+  would scale with unrelated posts.
+- Drop legacy aliases: rejected because canonicalization would hide retained
+  authorized history.

docs/adr/README.md

@@ -27,6 +27,7 @@ decision from them.
 | [`operability/http-concurrency-evidence.md`](../operability/http-concurrency-evidence.md) | [0204](0204-analysis-run-short-transaction-delivery.md), [0212](0212-single-query-authorized-post-filter-options.md), [0213](0213-global-ask-embedding-pool-release.md) |
 | [`operability/mcp-concurrency-evidence.md`](../operability/mcp-concurrency-evidence.md) | [0218](0218-current-contract-mcp-global-ask.md) |
 | Evidence operations Dashboard (`/`) | [0206](0206-evidence-operations-dashboard.md) |
+| Activity stream (`backend/app/activity_stream.py`) | [0363](0363-canonical-activity-stream-identity.md) |
 | [`temporal-topic-context-influence-research.md`](../temporal-topic-context-influence-research.md) | [0210](0210-temporal-topic-context-influence-dashboard.md) |
 | [`python-mathematical-compute-boundary-audit.md`](../doctoring/python-mathematical-compute-boundary-audit.md) | [0208](0208-externalize-local-mathematical-compute.md) |
 | [`WORKER_FUNCTION_TAXONOMY_REFERENCES.md`](../doctoring/WORKER_FUNCTION_TAXONOMY_REFERENCES.md) | [0232](0232-worker-function-taxonomy-in-the-published-ontology.md) |

tests/test_activity_stream.py

@@ -7,8 +7,15 @@
 
 from __future__ import annotations
 
+from inspect import signature
+
+from redis.exceptions import WatchError
+
 from backend.app.activity_stream import (
+    create_valkey_client,
+    publish_activity_event,
     publish_activity_event_sync,
+    read_activity_events,
     ticket_created_summary,
     ticket_status_changed_summary,
 )
@@ -17,34 +24,139 @@
 
 
 class _FakeStream:
+    """Small in-memory stand-in for the Valkey stream methods under contract."""
+
     def __init__(self) -> None:
+        """Start with no retained activity entries or key mutations."""
         self.entries: list[tuple[str, dict[str, str]]] = []
+        self.version = 0
 
-    def xrevrange(self, key: str, count: int = 50):
+    def xrevrange(self, key: str, count: int | None = None):
+        """Return newest-first entries with the same optional count boundary."""
         del key
-        return list(reversed(self.entries[-count:]))
+        entries = list(reversed(self.entries))
+        return entries if count is None else entries[:count]
 
     def xadd(self, key: str, fields: dict[str, str], maxlen=None, approximate=None):
+        """Append a copied wire record and advance the watched key version."""
         del key, maxlen, approximate
         entry_id = f"1-{len(self.entries)}"
         self.entries.append((entry_id, dict(fields)))
+        self.version += 1
         return entry_id
 
+    def pipeline(self):
+        """Create the WATCH/MULTI pipeline used by synchronous reseeding."""
+        return _FakePipeline(self)
+
+
+class _FakePipeline:
+    """Model the redis-py WATCH/MULTI behavior needed by the reseed contract."""
+
+    def __init__(self, stream: _FakeStream) -> None:
+        """Bind the transaction to one fake stream client."""
+        self.stream = stream
+        self.watched_version: int | None = None
+        self.pending_xadd: tuple[str, dict[str, str], object, object] | None = None
+        self.in_multi = False
+
+    def __enter__(self):
+        """Return this transaction context."""
+        return self
+
+    def __exit__(self, exc_type, exc, traceback) -> bool:
+        """Never suppress transaction exceptions."""
+        del exc_type, exc, traceback
+        return False
+
+    def watch(self, key: str) -> None:
+        """Remember the stream version at the optimistic-lock boundary."""
+        del key
+        self.watched_version = self.stream.version
+
+    def unwatch(self) -> None:
+        """Release the optimistic lock after finding an existing fact."""
+        self.watched_version = None
+
+    def xrevrange(self, key: str, count: int | None = None):
+        """Read immediately while the key is watched, as redis-py does."""
+        return self.stream.xrevrange(key, count=count)
+
+    def multi(self) -> None:
+        """Begin queuing the append for optimistic execution."""
+        self.in_multi = True
+
+    def xadd(self, key: str, fields: dict[str, str], maxlen=None, approximate=None):
+        """Queue one append after MULTI rather than mutating immediately."""
+        if not self.in_multi:
+            return self.stream.xadd(key, fields, maxlen=maxlen, approximate=approximate)
+        self.pending_xadd = (key, dict(fields), maxlen, approximate)
+        return self
+
+    def execute(self):
+        """Reject a stale watched snapshot or atomically apply the queued append."""
+        if self.watched_version != self.stream.version:
+            raise WatchError("synthetic concurrent stream mutation")
+        if self.pending_xadd is None:
+            return []
+        key, fields, maxlen, approximate = self.pending_xadd
+        return [self.stream.xadd(key, fields, maxlen=maxlen, approximate=approximate)]
+
+
+class _RaceInjectingStream(_FakeStream):
+    """Inject one concurrent seed after the caller reads its stale snapshot."""
+
+    def __init__(self, concurrent_fields: dict[str, str]) -> None:
+        """Remember the fact another seed execution will publish once."""
+        super().__init__()
+        self.concurrent_fields = dict(concurrent_fields)
+        self.injected = False
+
+    def xrevrange(self, key: str, count: int | None = None):
+        """Return the old snapshot, then simulate another process appending it."""
+        entries = super().xrevrange(key, count=count)
+        if not self.injected:
+            self.injected = True
+            super().xadd(key, self.concurrent_fields)
+        return entries
+
+
+def test_activity_stream_owned_parameters_use_semantic_names() -> None:
+    """Organization-owned activity helpers expose bounded-context vocabulary."""
+    assert list(signature(create_valkey_client).parameters) == ["valkey_url"]
+    expected_event_parameters = [
+        "valkey_client",
+        "post_id",
+        "event_type",
+        "actor_account_id",
+        "activity_summary",
+    ]
+    assert list(signature(publish_activity_event).parameters) == expected_event_parameters
+    assert list(signature(publish_activity_event_sync).parameters) == expected_event_parameters
+    assert list(signature(read_activity_events).parameters) == [
+        "valkey_client",
+        "post_id",
+        "event_count",
+    ]
+
 
 def test_ticket_created_summary_matches_the_live_api_wording() -> None:
+    """Ticket creation retains the wording consumed by the live Activity UI."""
     assert ticket_created_summary("Send Northridge Grid the revised quote") == (
         "Ticket created: Send Northridge Grid the revised quote"
     )
 
 
 def test_ticket_status_changed_summary_uses_the_lookup_label() -> None:
+    """Status activity uses the resolved label rather than leaking its raw code."""
     assert ticket_status_changed_summary("In progress") == (
         "Ticket status changed to In progress"
     )
     assert "in_progress" not in ticket_status_changed_summary("In progress")
 
 
-def test_publish_activity_event_sync_skips_a_matching_summary() -> None:
+def test_publish_activity_event_sync_skips_a_matching_activity_fact() -> None:
+    """An exact retained event-type, actor, and summary tuple is replay-idempotent."""
     client = _FakeStream()
     first = publish_activity_event_sync(
         client,
@@ -67,6 +179,140 @@ def test_publish_activity_event_sync_skips_a_matching_summary() -> None:
     assert "Send Northridge Grid the revised quote" in client.entries[0][1]["summary"]
 
 
+def test_publish_activity_event_sync_keeps_distinct_events_with_same_summary() -> None:
+    """Reseeding must not collapse different activity facts onto summary text."""
+    client = _FakeStream()
+    shared_summary = "Assignment updated"
+    client.xadd(
+        "activity:post-1",
+        {
+            "event_type": "ticket_status_changed",
+            "actor_account_id": "acct-2",
+            "summary": shared_summary,
+        },
+    )
+
+    created = publish_activity_event_sync(
+        client,
+        "post-1",
+        "ticket_created",
+        "acct-1",
+        shared_summary,
+    )
+
+    assert created == "1-1"
+    assert len(client.entries) == 2
+    assert client.entries[-1][1] == {
+        "event_type": "ticket_created",
+        "actor_account_id": "acct-1",
+        "summary": shared_summary,
+    }
+
+
+def test_publish_activity_event_sync_keeps_same_event_type_for_different_actor() -> None:
+    """Actor identity independently distinguishes otherwise equal reseed facts."""
+    client = _FakeStream()
+    shared_summary = "Assignment updated"
+    client.xadd(
+        "activity:post-1",
+        {
+            "event_type": "ticket_status_changed",
+            "actor_account_id": "acct-2",
+            "summary": shared_summary,
+        },
+    )
+
+    created = publish_activity_event_sync(
+        client,
+        "post-1",
+        "ticket_status_changed",
+        "acct-1",
+        shared_summary,
+    )
+
+    assert created == "1-1"
+    assert client.entries[-1][1]["actor_account_id"] == "acct-1"
+
+
+def test_publish_activity_event_sync_keeps_same_actor_for_different_event_type() -> None:
+    """Event type independently distinguishes otherwise equal reseed facts."""
+    client = _FakeStream()
+    shared_summary = "Assignment updated"
+    client.xadd(
+        "activity:post-1",
+        {
+            "event_type": "ticket_status_changed",
+            "actor_account_id": "acct-1",
+            "summary": shared_summary,
+        },
+    )
+
+    created = publish_activity_event_sync(
+        client,
+        "post-1",
+        "ticket_created",
+        "acct-1",
+        shared_summary,
+    )
+
+    assert created == "1-1"
+    assert client.entries[-1][1]["event_type"] == "ticket_created"
+
+
+def test_publish_activity_event_sync_scans_the_retained_stream_for_reseed_idempotency() -> None:
+    """A retained seed event stays idempotent after more than 50 newer events."""
+    client = _FakeStream()
+    seed_summary = ticket_created_summary("Send Northridge Grid the revised quote")
+    assert publish_activity_event_sync(
+        client,
+        "post-1",
+        "ticket_created",
+        "acct-1",
+        seed_summary,
+    ) == "1-0"
+
+    for index in range(51):
+        client.xadd(
+            "activity:post-1",
+            {
+                "event_type": "ticket_status_changed",
+                "actor_account_id": "acct-1",
+                "summary": f"newer activity {index}",
+            },
+        )
+
+    entry_count = len(client.entries)
+    assert publish_activity_event_sync(
+        client,
+        "post-1",
+        "ticket_created",
+        "acct-1",
+        seed_summary,
+    ) is None
+    assert len(client.entries) == entry_count
+
+
+def test_publish_activity_event_sync_retries_when_another_seed_wins_the_race() -> None:
+    """Concurrent make-seed executions must not append the same activity twice."""
+    expected_fields = {
+        "event_type": "ticket_created",
+        "actor_account_id": "acct-1",
+        "summary": ticket_created_summary("Send Northridge Grid the revised quote"),
+    }
+    client = _RaceInjectingStream(expected_fields)
+
+    created = publish_activity_event_sync(
+        client,
+        "post-1",
+        expected_fields["event_type"],
+        expected_fields["actor_account_id"],
+        expected_fields["summary"],
+    )
+
+    assert created is None
+    assert client.entries == [("1-0", expected_fields)]
+
+
 def test_valkey_child_span_shares_parent_trace_id(monkeypatch) -> None:
     """Same-process Valkey work inherits the parent TraceId."""
     from opentelemetry import trace
@@ -75,7 +321,10 @@ def test_valkey_child_span_shares_parent_trace_id(monkeypatch) -> None:
     captured: dict[str, str] = {}
 
     class _Client(_FakeStream):
+        """Capture tracing identity at the fake Valkey append boundary."""
+
         def xadd(self, key: str, fields: dict[str, str], maxlen=None, approximate=None):
+            """Record the current child span before delegating to the fake stream."""
             span = trace.get_current_span()
             captured["trace_id"] = format(span.get_span_context().trace_id, "032x")
             captured["span_id"] = format(span.get_span_context().span_id, "016x")