Stream-driven live updates, shared freshness reads and ETag/304 for console reads

CHANGELOG.md

@@ -18,6 +18,7 @@ the project uses [semantic versioning](https://semver.org/) once it reaches 1.0;
 
 ### Changed
 
+- **Live updates no longer poll every 5 seconds.** The console refreshes from the live-update stream and on focus, and polls (with jitter) only while the stream is down. Web processes share one freshness read per organization (at most about one per 400 ms), and stream lifetimes are randomized (40 to 55 seconds) so browsers do not reconnect together. `task-views`, `decisions` and `notifications` reads send `ETag` and answer `304` when unchanged; responses stay `no-store`.
 - **Console navigation is reorganised by job.** The sidebar groups daily work (**Home**, **Inbox**, **Tasks**, **Chat**), **Agents** (**Agents**, **Find tasks**, **Orchestration**) and **You** (**Notifications**, **Settings**, **Administration** for administrators).
 - **Administration is its own area** (`/admin`) with **Overview**, **People**, **Teams**, **Agent access** and **Policies & alerts**, plus a link to the audit trail. It replaces the organization cards that sat under **Settings**, which now holds only your browser-stored request defaults and extensions. **Teams** gains **Disable team**. No API changed.
 - **People management in the console.** With OIDC sign-in, **Administration → People** can add a person, change a role, deactivate and reactivate, all audited (`member.*`) and with the same guards as the API: you cannot change or deactivate yourself. In local development mode the page shows the built-in administrator and explains that people need OIDC. `POST /api/admin/security` gains `provisionMember`, `changeRole`, `deactivateMember` and `reactivateMember`; `GET` adds `identity` and `inactiveMemberships`.

docs/concepts/durable-tasks.md

@@ -103,7 +103,7 @@ Raw events live in an append-only ledger. What the screens read is a projection:
 
 The projection is derived by a deterministic reducer, so it can be rebuilt from retained events. Rebuild prepares the new generation outside the task lock, verifies archived binary digests, then swaps it in only if the task has not changed in the meantime. A failed rebuild keeps the old generation readable. Rebuild never sends anything to an agent. It cannot restore history that retention has deleted. See [Upgrading and backups](../guides/upgrading-and-backups.md) for the command.
 
-To tell browsers something changed, every accepted projection write queues a `task.freshness` outbox row. A worker publishes it. `GET /api/tasks/events` streams three content-free events: `ready` (sent on every connection), `freshness` and `resync` (after about 15 s of quiet). Connections last about 55 s and then the browser reconnects. Signals carry no task data. On any signal, on reconnect, on focus and on a 5-second fallback poll, the browser re-queries. A lost signal costs at most a few seconds of staleness.
+To tell browsers something changed, every accepted projection write queues a `task.freshness` outbox row. A worker publishes it. `GET /api/tasks/events` streams three content-free events: `ready` (sent on every connection), `freshness` and `resync` (after about 15 s of quiet). Connections last 40 to 55 s (randomized) and then the browser reconnects. Signals carry no task data. On any signal, on reconnect and on focus, the browser re-queries. It polls (about every 5–7.5 seconds, with jitter) only while the stream is down: when the browser has no `EventSource`, or no frame has arrived for 20 seconds or a reconnect has failed for 5. The stream stays open for 40 to 55 seconds at a time, so an idle tab makes one re-query per signal, about one every 15 seconds, and no timer reads while the stream is healthy. Re-queries send `If-None-Match` from tab memory and get `304` when nothing changed, so an unchanged read costs no response body and no re-render.
 
 ## Identity of remote IDs
 
@@ -118,7 +118,7 @@ Known `referenceTaskIds` on user and agent messages create durable local links w
 | Command stuck `pending` | No worker running, or the agent is disabled | Check the worker is running. See [Running and workers](../operations/running-and-workers.md) |
 | Command `uncertain` | Failure after dispatch, or lease expiry | Open the task, check its state, and send again only if the agent did nothing |
 | Task never leaves `working` with no browser open | Agent lacks streaming and push is off | Reconciliation will catch up within about 15 s. Check agent health |
-| UI is a few seconds stale | Freshness signal lost | The 5 s fallback poll covers it |
+| UI is a few seconds stale | Freshness signal lost | The next `resync` (within 15 s) covers it, or the fallback poll if the stream is down |
 | Rebuild refuses | Ledger empty or an archive is missing or corrupt | Restore from backup. It fails closed on purpose |
 
 More in [Troubleshooting](../operations/troubleshooting.md).

docs/guides/production-deployment.md

@@ -126,7 +126,8 @@ Run both processes under a supervisor that restarts them. A crash loses nothing
 Terminate TLS at a proxy and forward to the web process. Requirements specific to Taskbay:
 
 - Forward everything under `/` to the web process. Agents call `POST /api/webhooks/a2a/<registration id>` without a browser session; do not put your own authentication in front of that path if you enable push.
-- Do not buffer or time out `text/event-stream` responses. The live-update stream (`GET /api/tasks/events`) ends by itself after 55 seconds and the browser reconnects; it already sets `X-Accel-Buffering: no`.
+- Do not buffer or time out `text/event-stream` responses. The live-update stream (`GET /api/tasks/events`) ends by itself after 40 to 55 seconds and the browser reconnects; it already sets `X-Accel-Buffering: no`.
+- Several web replicas (for example on Kubernetes) need no sticky sessions: a stream is stateless, lives 40 to 55 seconds and reconnects to whichever replica the balancer picks. Freshness travels through the database, so a change committed by any replica or worker wakes browsers connected to every other replica. Each replica reads the freshness token at most about 2.5 times a second per organization that has an open stream, however many browsers are connected. Set the ingress or load-balancer idle timeout above 60 seconds, and let a draining replica close its streams; browsers reconnect on their own.
 - The request body limit in the application is 16 MB.
 - The public origin in `A2A_AUTH_ORIGIN` must be exactly what browsers use. The login redirect and the `Origin` check for mutations come from it, not from `Host` headers.
 

docs/guides/use-the-console.md

@@ -370,7 +370,7 @@ Administrators also see a link to **Administration** at the bottom of the page.
 | A form shows as raw data | The agent does not advertise the extension, the form is outside the supported subset, or it was already answered | See [Rich content](../concepts/rich-content.md#when-it-fails) |
 | "Conversation not found" / "This conversation is unavailable in the current organization." | The link is from another organization, or you lost read access | Open the chat from **Recent conversations**, or ask for a grant |
 | "Task not found" | The task does not exist or you cannot read it. The server answers 404 rather than 403. | Ask an administrator for a grant |
-| Banner ending "Retrying server sync…" | The browser could not load tasks from the server | Check the server is running. The console retries every 5 seconds. |
+| Banner ending "Retrying server sync…" | The browser could not load tasks from the server | Check the server is running. The console retries on the next live-update signal (about every 15 seconds), or about every 5–7 seconds while the live-update stream is down. |
 | "Administrators only" on **Audit trail** | You are not an administrator | Open a task and use **View the full trail** |
 | **Administration** missing from the sidebar, or "Administrators only" | You are not an administrator | Ask an administrator |
 | "Too many requests. Try again shortly." | Per-person rate limit | Wait a minute |

docs/operations/running-and-workers.md

@@ -75,9 +75,9 @@ All loops are idempotent and retry after a database error, logging a single line
 | Approvals and escalation | Expires overdue approval requests, marks requests superseded when their task finished, refreshes in-flight approved deliveries, and escalates overdue owned tasks. | Every `A2A_DECISION_SWEEP_MS` (default 15000; values below 100, or not a number, fall back to 15000). |
 | Notifications | Turns events into inbox rows, then sends each to the signed webhook if one is configured. Webhook delivery retries up to 8 times with backoff from 5 s to 10 min. | Polls every 250 ms. |
 
-The browser side: `GET /api/tasks/events` is a server-sent-events stream that sends `ready`, then `freshness` when something committed and `resync` about every 15 seconds. It closes itself after 55 seconds
-and the browser reconnects. It carries no task content; the browser re-reads state from the API. With PostgreSQL, web replicas poll a freshness token every 500 ms, so a change made by a worker on another host
-shows up without a message broker.
+The browser side: `GET /api/tasks/events` is a server-sent-events stream that sends `ready`, then `freshness` when something committed and `resync` about every 15 seconds. It closes itself after 40 to 55 seconds (randomized)
+and the browser reconnects. It carries no task content; the browser re-reads state from the API. With PostgreSQL, each web process polls the freshness token every 500 ms while a stream is open, so a change made by a worker on another host
+shows up without a message broker. Open streams in a process share those reads: the database sees at most one token read per organization per 400 ms per web process, however many browsers are connected.
 
 ## Scaling and leases
 

docs/reference/http-api.md

@@ -52,7 +52,7 @@ copied from a browser. That is an awkward workflow, and the reason a token API i
 | Topic | Behavior |
 | --- | --- |
 | Format | JSON in, JSON out, `Content-Type: application/json`. Request bodies larger than 16 MiB are rejected with `400`. A missing body is `400 Request body is required.`; invalid JSON is `400 Invalid request JSON.` |
-| Caching | Responses carry `Cache-Control: no-store`. |
+| Caching | Responses carry `Cache-Control: no-store`. `GET /api/task-views`, `GET /api/decisions` and `GET /api/notifications` also carry a weak `ETag` of the body and answer `304` (no body, still `no-store`) to a matching `If-None-Match`. The console sends it from tab memory, so the browser stores nothing. A `304` saves the response body and the client re-render; the server still runs the query. |
 | IDs | Local IDs are UUIDs generated by Agent Taskbay. A task has a local ID (`localId`, used in URLs) and a remote ID (`taskId` from the agent); they are different. Remote IDs are only unique within `(agent, tenant)`. |
 | Timestamps | ISO 8601 UTC strings. |
 | Idempotency | Mutations that start durable work take an `Idempotency-Key` header (1 to 255 characters). Reusing a key with identical content returns the original result; with different content it returns `409`. |
@@ -458,7 +458,7 @@ A message that reached Taskbay only because the agent's history was loaded has `
 
 `read`. A long-lived SSE signal that "something changed for this organization". Frames carry no task data (`data: {}`); clients re-query. Event names:
 `ready` (first frame), `freshness` (a change was committed), `resync` (sent after 15 seconds of silence so clients refresh anyway). Each frame is preceded by
-`retry: 1000`. The server closes the stream after 55 seconds; reconnect.
+`retry: 1000`. The server closes the stream after 40 to 55 seconds (randomized so connections do not expire together); reconnect.
 
 ## Task workflow
 

scripts/verify-task-http.mjs

@@ -394,7 +394,13 @@ try {
   assert.deepEqual((await json("/api/task-views?organizationId=foreign&tenant=foreign")).tasks, browserOne);
   assert.equal((await fetch(base + "/api/task-views?limit=0")).status, 400);
   assert.equal((await fetch(base + "/api/task-views?after=" + remoteId)).status, 400);
-  assert.equal((await fetch(base + "/api/task-views")).headers.get("cache-control"), "no-store");
+  const viewsResponse = await fetch(base + "/api/task-views");
+  assert.equal(viewsResponse.headers.get("cache-control"), "no-store");
+  const viewsTag = viewsResponse.headers.get("etag");
+  assert.ok(viewsTag);
+  const unchangedViews = await fetch(base + "/api/task-views", { headers: { "If-None-Match": viewsTag } });
+  assert.equal(unchangedViews.status, 304);
+  assert.equal(unchangedViews.headers.get("cache-control"), "no-store");
   await json(`/api/agents/${one.id}/tasks/${remoteId}/cancel?tenant=tenant-b`, { method: "POST" });
   assert.equal((await json(`/api/tasks/${ids[2]}`)).task.state, "TASK_STATE_CANCELED");
   assert.equal((await json("/api/task-views")).tasks.find((task) => task.localId === ids[2]).state, "TASK_STATE_CANCELED");

src/app/api/decisions/route.ts

@@ -1,4 +1,5 @@
 import { authenticatedRoute } from "@/server/runtime/identity";
+import { conditionalJson } from "@/lib/conditional-json";
 import { apiError } from "@/lib/api-response";
 import { readJsonRequest } from "@/lib/request-guard";
 import { createDecisionService, listDecisions, openDecisionSchema, parseBody, toAction, requestView, requirePrincipal } from "@/server/runtime/decisions";
@@ -14,7 +15,7 @@ async function handleGET(request: Request) {
     if (status && !(STATUSES as readonly string[]).includes(status)) return Response.json({ error: { message: "Unknown status." } }, { status: 400 });
     const requests = await listDecisions({ status: (status ?? undefined) as (typeof STATUSES)[number] | undefined,
       taskId: params.get("taskId") ?? undefined, mine: params.get("assigned") === "me" });
-    return Response.json({ decisions: requests }, { headers: { "Cache-Control": "no-store" } });
+    return conditionalJson(request, { decisions: requests });
   } catch (error) { return apiError(error, 400); }
 }
 

src/app/api/notifications/route.ts

@@ -1,4 +1,5 @@
 import { authenticatedRoute } from "@/server/runtime/identity";
+import { conditionalJson } from "@/lib/conditional-json";
 import { apiError } from "@/lib/api-response";
 import { inboxQuerySchema, listInbox } from "@/server/runtime/notifications";
 export const runtime = "nodejs";
@@ -8,7 +9,7 @@ async function handleGET(request: Request) {
   try {
     const parsed = inboxQuerySchema.safeParse(Object.fromEntries(new URL(request.url).searchParams));
     if (!parsed.success) return Response.json({ error: { message: parsed.error.issues.map((issue) => issue.message).join("; ") } }, { status: 400 });
-    return Response.json(await listInbox({ unreadOnly: parsed.data.unread === "true", limit: parsed.data.limit, cursor: parsed.data.cursor }), { headers: { "Cache-Control": "no-store" } });
+    return conditionalJson(request, await listInbox({ unreadOnly: parsed.data.unread === "true", limit: parsed.data.limit, cursor: parsed.data.cursor }));
   } catch (error) { return apiError(error, 400); }
 }
 

src/app/api/task-views/route.ts

@@ -1,4 +1,5 @@
 import { authenticatedRoute } from "@/server/runtime/identity";
+import { conditionalJson } from "@/lib/conditional-json";
 import { apiError } from "@/lib/api-response";
 import { withTaskQueries } from "@/server/runtime/task-persistence";
 
@@ -16,7 +17,7 @@ async function handleGET(request: Request) {
       return Response.json({ error: { message: "limit must be 1–100 and after must be a local UUID." } }, { status: 400 });
     }
     const page = await withTaskQueries((queries, organizationId) => queries.contentPage(organizationId, limit, after));
-    return Response.json(page, { headers: { "Cache-Control": "no-store" } });
+    return conditionalJson(request, page);
   } catch (error) { return apiError(error, 500); }
 }
 

src/app/api/tasks/events/route.ts

@@ -1,7 +1,7 @@
 import { authenticatedRoute } from "@/server/runtime/identity";
 import { apiError } from "@/lib/api-response";
 import { withTaskQueries } from "@/server/runtime/task-persistence";
-import { runtimeFreshness } from "@/server/runtime/freshness";
+import { liveFreshnessReader } from "@/server/runtime/freshness";
 import { freshnessStream } from "@/server/adapters/live/freshness-stream";
 
 export const runtime = "nodejs";
@@ -11,7 +11,7 @@ async function handleGET(request: Request) {
   try {
     // Like task reads, scope comes from the server, never a query/header selector.
     const organizationId = await withTaskQueries(async (_queries, organization) => organization);
-    const reader = runtimeFreshness();
+    const reader = liveFreshnessReader();
     await reader.readToken(organizationId); // Fail before returning streaming headers.
     return new Response(freshnessStream(reader, organizationId, request.signal), {
       headers: { "Content-Type": "text/event-stream", "Cache-Control": "no-store, no-transform",

src/components/chat/durable-task-sync.tsx

@@ -10,7 +10,6 @@ import { useTaskStore } from "@/store/task-store";
 export function DurableTaskSync() {
   useEffect(() => {
     const controller = new AbortController();
-    let timer: ReturnType<typeof setTimeout>;
     const load = queuedRead(async () => {
       const revision = useTaskStore.getState().revision;
       try {
@@ -20,15 +19,11 @@ export function DurableTaskSync() {
         if (!controller.signal.aborted) useTaskStore.getState().fail(error instanceof Error ? error.message : "Could not load server tasks.");
       }
     }, controller.signal);
-    const poll = async () => {
-      await load();
-      if (!controller.signal.aborted) timer = setTimeout(poll, 5000);
-    };
-    void poll();
+    void load();
     const onFocus = () => { void load(); };
     const unsubscribe = subscribeTaskFreshness(onFocus);
     window.addEventListener("focus", onFocus);
-    return () => { controller.abort(); clearTimeout(timer); unsubscribe(); window.removeEventListener("focus", onFocus); };
+    return () => { controller.abort(); unsubscribe(); window.removeEventListener("focus", onFocus); };
   }, []);
   return null;
 }

src/lib/conditional-fetch.ts

@@ -0,0 +1,21 @@
+const MAX_ENTRIES = 50;
+const held = new Map<string, { etag: string; body: unknown }>();
+
+/**
+ * GET JSON with `If-None-Match` from a tab-local memory cache. Responses are `no-store`, so the browser
+ * keeps nothing; a `304` reuses the body held here. `unchanged` lets callers skip re-rendering.
+ */
+export async function fetchJson<T>(url: string, signal?: AbortSignal) {
+  const previous = held.get(url);
+  const response = await fetch(url, { cache: "no-store", signal,
+    ...(previous ? { headers: { "If-None-Match": previous.etag } } : {}) });
+  if (response.status === 304 && previous) return { ok: true as const, body: previous.body as T, unchanged: true };
+  const body = await response.json();
+  const etag = response.headers.get("etag");
+  if (response.ok && etag) {
+    held.delete(url);
+    held.set(url, { etag, body });
+    if (held.size > MAX_ENTRIES) held.delete(held.keys().next().value as string);
+  } else held.delete(url);
+  return { ok: response.ok, body: body as T, unchanged: false };
+}