Observe sessions and recorded usage in the Janitor dashboard

.scratch/multiplayer-janitor-implementation/issues/04-session-observation.md

@@ -4,14 +4,24 @@
 
 **Blocked by:** 03: Collaborate with Janitor in a private Slack thread.
 
-**Status:** ready-for-agent
+**Status:** needs-triage
 
 Source: [Accepted implementation handoff](../spec.md). Implement this slice within its accepted contracts and evidence limits.
 
-- [ ] Provide paginated session list/detail with title, repository when known, home-thread and PR associations, meaningful activity, execution reason, usage, delivery warnings and freshness.
-- [ ] Gate HTTP reads and live subscriptions on valid browser Access and active Janitor membership. Admin/member visibility is equivalent and independent of Slack channel membership; removal prevents continued subscription access.
-- [ ] Project working/idle/blocked/failed from durable facts, independently of delivery health and projection staleness. Completed turns return to idle; historical failures do not override newer outcomes.
-- [ ] Apply cumulative usage snapshots and exclusive cursor advancement atomically by replacement. Display input plus cache read/write and output plus reasoning; distinguish unavailable totals and explain omitted/unreported usage.
-- [ ] Order working sessions first then meaningful activity with stable pagination. Heartbeats, usage-only changes and delivery retry bookkeeping do not reorder activity.
-- [ ] Commit invalidation intent with projection updates and reuse HTTP refresh/live invalidation/reconnect/fallback behavior. Preserve last known data with a stale/reconnecting indication.
-- [ ] Test replay, sequence gaps, missing invalidations, late updates, reconnects, removed viewers and successful work with pending delivery. Do not add logs, conversation history, spending or team totals.
+- [x] Provide paginated session list/detail with title, repository when known, home-thread and PR associations, meaningful activity, execution reason, usage, delivery warnings and freshness.
+- [x] Gate HTTP reads and live subscriptions on valid browser Access and active Janitor membership. Admin/member visibility is equivalent and independent of Slack channel membership; removal prevents continued subscription access.
+- [x] Project working/idle/blocked/failed from durable facts, independently of delivery health and projection staleness. Completed turns return to idle; historical failures do not override newer outcomes.
+- [x] Apply cumulative usage snapshots and exclusive cursor advancement atomically by replacement. Display input plus cache read/write and output plus reasoning; distinguish unavailable totals and explain omitted/unreported usage.
+- [x] Order working sessions first then meaningful activity with stable pagination. Heartbeats, usage-only changes and delivery retry bookkeeping do not reorder activity.
+- [x] Commit invalidation intent with projection updates and reuse HTTP refresh/live invalidation/reconnect/fallback behavior. Preserve last known data with a stale/reconnecting indication.
+- [x] Test replay, sequence gaps, missing invalidations, late updates, reconnects, removed viewers and successful work with pending delivery. Do not add logs, conversation history, spending or team totals.
+
+## Comments
+
+2026-09-13: Implemented for maintainer review. Migration `0028_session_observation.sql` adds the team-wide `sessions` live channel with invalidation triggers on the session projection, session identity, home-thread associations, delivery health and membership. `SessionObservation` composes one read model from Janitor's session identity and associations, the runner-projected execution facts and usage, and the delivery records: paginated list ordered working first, then meaningful activity, then session ID, plus compact detail with pending inputs, latest error, pending delivery and recovery status. Cumulative usage is replaced under cursor protection and displayed as input plus cache read/write and output plus reasoning, unavailable until the first snapshot. Accepted input is meaningful activity and moves an idle or failed session to working; runner holds are blocked with their reason; delivery warnings and projection freshness stay independent of execution state.
+
+Routes under `/api/v1/sessions` sit behind Access and active membership; the live channel rechecks membership at the boundary and removal closes open subscriptions through the membership topic. The web dashboard at `/sessions` reuses the repository live pattern (socket, reconnect refresh, fallback timer), coalesces invalidations during a read, and keeps last known data with a stale or reconnecting indication when a refresh fails. No conversation history, logs, spending or team totals are exposed.
+
+Tests cover ordering and stable pages under usage-only changes, unavailable versus normalized-zero usage and out-of-order snapshot rejection, execution/delivery/freshness independence, turn completion returning to idle, new input superseding a historical failure, pending threads and hidden disconnected sessions, invalidation intent rolling back with the projection, membership revocation, route authorization and the browser refresh/stale/denied behaviour.
+
+Validation: `vp check` passed with warnings and no errors; the full suite passed (backend and web) with the runner-backed suites skipped as before. The review found no documented-standard breaches and one P1 specification issue (a terminally rejected admission left the session working), fixed with a regression test, plus four P2 findings (freshness read failures not invalidating, invalidation refreshes dropping loaded pages, stale rejection errors on idle sessions, untested reconnect and fallback paths) and the platform-recovery scope creep, all resolved. See [the review record](../review-04.md).

.scratch/multiplayer-janitor-implementation/review-04.md

@@ -0,0 +1,28 @@
+# Ticket 04 implementation review
+
+Baseline: `3774605` (the working tree before commit). Two independent reviewers examined the implementation: one against the repository's documented standards and Fowler's smell baseline, one against the ticket, the accepted specification's "Observation and lifecycle" section and the observation contracts decision.
+
+## Standards
+
+No documented-standard breaches. Judgement calls acted on: the Slack permalink helper is now shared (`Slack/HomeThread.ts`), the live-route expiry bound is shared (`Ingress/Live.ts`), the web read-failure decoding is shared (`lib/api.ts`), the unused domain topic constant and the wire usage buckets were removed, the platform label switch was extracted, and `activity_key`/`count` were renamed. Left as is: the dashboard navigation says "Sessions" rather than the glossary's "Janitor dashboard"; the projection table now has more than one writer (input acceptance and terminal admission rejection mark the session working or failed before the runner's events arrive), which the spec's "accepted work is being admitted" state requires. Glossary gaps to record with `domain-modeling`: execution states, meaningful activity, projection freshness, invalidation intent.
+
+## Spec
+
+Findings and outcomes:
+
+1. P1: a terminally rejected admission left the session working with "input pending" indefinitely. The rejection now marks the projection failed with the rejection reason; a later admission returns it to working. Regression asserted in the handoff test.
+2. P2: read failures updated only the catch-up obligation and never invalidated, while the projection's heartbeat did. The catch-up obligation's last read and error now notify; the heartbeat column no longer does.
+3. P2: an invalidation refresh re-read only the first page. A refresh now re-reads as many rows as are on screen (server cap 100).
+4. P2: an old rejected input's error stayed on idle sessions with newer work. Only the latest accepted input's rejection is shown.
+5. P2: no browser test for reconnect or fallback refresh. Added, with page preservation and an unrelated-topic check.
+6. P3: platform recovery status in detail belonged to ticket 10 and was removed; input counts stay as concise activity; the misleading ordering index was dropped from the migration.
+
+Verified correct by the reviewer: keyset ordering and pagination, usage replacement under cursor protection, unavailable versus normalized zero, membership recheck at the subscription boundary with revocation closing sockets, HTTP reads behind membership, and removal flushing invalidations.
+
+## Validation
+
+- `vp check` passed with warnings and no errors.
+- The full suite passed after the fixes; the runner-backed suites stay skipped because the runner's isolated dependencies are not installed.
+- No live Slack, GitHub, runner or deployment was exercised.
+
+Standards: zero breaches, smells addressed where concrete. Spec: six findings resolved, highest severity P1; zero outstanding.

apps/cluster/migrations/0028_session_observation.sql

@@ -0,0 +1,30 @@
+-- Session observation: one team-wide live channel for the dashboard.
+--
+-- Session projections are not owned by a repository, so their invalidation
+-- intent is keyed by the reserved channel 'sessions' rather than a repository
+-- id. Rows commit with the change that caused them; the dispatcher forwards
+-- them after commit and clients refresh their HTTP reads. Membership changes
+-- use the same channel so an open subscription of a removed teammate closes.
+CREATE FUNCTION enqueue_session_notification() RETURNS trigger LANGUAGE plpgsql AS $$
+BEGIN
+  -- The projection's freshness_at is a heartbeat; the dashboard shows the
+  -- catch-up obligation's last read instead, which has its own trigger.
+  IF TG_OP = 'UPDATE' AND (to_jsonb(OLD) - 'freshness_at') = (to_jsonb(NEW) - 'freshness_at') THEN RETURN NULL; END IF;
+  INSERT INTO live_notification(repository_id,topic) VALUES('sessions',TG_ARGV[0])
+    ON CONFLICT(repository_id,topic) DO UPDATE SET revision=nextval('live_notification_revision');
+  RETURN NULL;
+END $$;
+
+CREATE TRIGGER live_session_projection AFTER INSERT OR UPDATE OR DELETE ON agent_session_projection
+  FOR EACH ROW EXECUTE FUNCTION enqueue_session_notification('sessions');
+CREATE TRIGGER live_session_freshness AFTER UPDATE OF last_read_at, last_error ON agent_catchup
+  FOR EACH ROW EXECUTE FUNCTION enqueue_session_notification('sessions');
+CREATE TRIGGER live_session AFTER INSERT OR UPDATE OF title, repository_id, runner_state, runner_error OR DELETE ON agent_session
+  FOR EACH ROW EXECUTE FUNCTION enqueue_session_notification('sessions');
+-- Delivery health and associations live on the thread; cursors and leases are bookkeeping.
+CREATE TRIGGER live_session_thread AFTER INSERT OR UPDATE OF repository_id, pr_number, state, warning, delivery_warning OR DELETE ON slack_thread
+  FOR EACH ROW EXECUTE FUNCTION enqueue_session_notification('sessions');
+CREATE TRIGGER live_session_github_output AFTER INSERT OR UPDATE OF state, error OR DELETE ON github_feedback_output
+  FOR EACH ROW EXECUTE FUNCTION enqueue_session_notification('sessions');
+CREATE TRIGGER live_membership AFTER UPDATE OF status ON teammate
+  FOR EACH ROW EXECUTE FUNCTION enqueue_session_notification('membership');

apps/cluster/migrations/README.md

@@ -219,3 +219,16 @@ events. Existing tables are unchanged.
 session. The identity stays available for external cleanup after repository
 removal. Session creation rejects a retry that changes the selection. Repository
 execution checks readiness through Janitor before cloning or dispatching tools.
+
+## Session observation
+
+`0028_session_observation.sql` adds the team-wide `sessions` live channel. It
+is keyed like a repository in `live_notification` but is not one: the
+dispatcher never marks it disconnected, and its `membership` topic carries the
+removed teammates whose open subscriptions must close. Triggers on the session
+projection, session identity, home-thread associations, delivery health and the
+catch-up obligation's last read commit invalidation intent with the change that
+caused it; catch-up reads and delivery cron wakes then forward it. The
+projection's own `freshness_at` heartbeat, thread cursors and leases are
+bookkeeping and do not notify. No data changes; existing sessions appear on the dashboard
+once their next projection read commits.

apps/cluster/src/Agent/CatchUpCron.ts

@@ -9,6 +9,7 @@ import * as Clock from "effect/Clock"
 import * as Effect from "effect/Effect"
 import * as Singleton from "effect/unstable/cluster/Singleton"
 import * as SqlClient from "effect/unstable/sql/SqlClient"
+import { flushLive } from "../LiveUpdates.ts"
 import { WorkflowDispatcher } from "../WorkflowDispatcher.ts"
 import { WorkflowOutbox } from "../WorkflowOutbox.ts"
 import { AgentEventProjection } from "./EventProjection.ts"
@@ -62,11 +63,14 @@ export const AgentCatchUpCronLayer = Singleton.make(
     }
     for (;;) {
       const summaries = yield* projection.processDue(50)
-      if (summaries.length > 0)
+      if (summaries.length > 0) {
         yield* Effect.logInfo("Caught up agent sessions", {
           sessions: summaries.length,
           applied: summaries.reduce((sum, summary) => sum + summary.applied, 0),
         })
+        // Projection changes committed their invalidation intent; forward it now.
+        yield* flushLive
+      }
       const next = yield* projection.nextDueIn
       const remaining = WAKE_BUDGET_MS - ((yield* Clock.currentTimeMillis) - started)
       if (next === null || next > remaining) break

apps/cluster/src/Agent/EventProjection.ts

@@ -160,7 +160,8 @@ export class AgentEventProjection extends Context.Service<
               activity = event.created
               break
             case "session.inbox.enqueued":
-              if (execution === "idle") {
+              // New accepted work supersedes a historical failure.
+              if (execution === "idle" || execution === "failed") {
                 execution = "working"
                 reason = "input pending"
               }

apps/cluster/src/Agent/Handoff.ts

@@ -206,6 +206,12 @@ export const deliverSession = Effect.fn("AgentHandoff.deliverSession")(function*
           UPDATE agent_input SET handoff_state = 'rejected', handoff_error = ${failure.message}
           WHERE session_id = ${sessionId} AND sequence = ${next.sequence}
         `)
+        // Admission reached a terminal failure: the dashboard must not keep showing
+        // the input as pending. A later admission moves the session back to working.
+        yield* query(sql`
+          UPDATE agent_session_projection SET execution = 'failed', reason = ${`Input rejected: ${failure.message}`}
+          WHERE session_id = ${sessionId} AND execution = 'working' AND reason = 'input pending'
+        `)
         continue
       }
       yield* query(sql`