1import type {
2 EnvironmentId,
3 OrchestrationV2DomainEvent,
4 OrchestrationV2ThreadShell,
5 OrchestrationV2TurnItem,
6 Project,
7 ThreadId,
8} from "@t3tools/contracts";
9import {
10 RelayApi,
11 type RelayAgentActivityPublishProofPayload,
12 type RelayAgentActivityState,
13} from "@t3tools/contracts/relay";
14import { projectThreadAwarenessV2 } from "@t3tools/shared/agentAwareness";
15import { turnItemUpdateCanEndBackgroundWork } from "@t3tools/shared/orchestrationV2PendingBackgroundWork";
16import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker";
17import {
18 normalizeRelayIssuer,
19 RELAY_ACTIVITY_PUBLISH_TYP,
20 signRelayJwt,
21} from "@t3tools/shared/relayJwt";
22import * as Cause from "effect/Cause";
23import * as Context from "effect/Context";
24import * as Crypto from "effect/Crypto";
25import * as DateTime from "effect/DateTime";
26import * as Effect from "effect/Effect";
27import * as Fiber from "effect/Fiber";
28import * as Layer from "effect/Layer";
29import * as Option from "effect/Option";
30import * as Queue from "effect/Queue";
31import * as Ref from "effect/Ref";
32import type * as Scope from "effect/Scope";
33import * as Stream from "effect/Stream";
34import * as FetchHttpClient from "effect/http/FetchHttpClient";
35import * as HttpClient from "effect/http/HttpClient";
36import * as HttpClientRequest from "effect/http/HttpClientRequest";
37import * as HttpApiClient from "effect/http-api/HttpApiClient";
38
39import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
40import {
41 isAgentActivityPublishingEnabledValue,
42 PUBLISH_AGENT_ACTIVITY_SECRET,
43 RELAY_ENVIRONMENT_CREDENTIAL_SECRET,
44 RELAY_ISSUER_SECRET,
45 RELAY_URL_SECRET,
46} from "../cloud/config.ts";
47import { getOrCreateEnvironmentKeyPairFromSecretStore } from "../cloud/environmentKeys.ts";
48import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
49import * as ThreadManagement from "../orchestration-v2/ThreadManagementService.ts";
50import * as ProjectService from "../project/ProjectService.ts";
51import { forkParked } from "../serverActivation.ts";
52
53export class AgentAwarenessRelay extends Context.Service<
54 AgentAwarenessRelay,
55 {
56 readonly publishThread: (threadId: ThreadId) => Effect.Effect<void>;
57 readonly drain: Effect.Effect<void>;
58 /** Retries a pending catch-up publish now. Call after this process links or enables publishing. */
59 readonly requestCatchUp: () => Effect.Effect<void>;
60 readonly start: () => Effect.Effect<void, never, Scope.Scope>;
61 }
62>()("t3/relay/AgentAwarenessRelay") {}
63
64function eventThreadId(event: OrchestrationV2DomainEvent): ThreadId {
65 return event.threadId;
66}
67
68// The filter takes loosely typed events; a turn-item payload carries both fields.
69function isTurnItemPayload(
70 payload: unknown,
71): payload is Pick<OrchestrationV2TurnItem, "type" | "status"> {
72 return (
73 typeof payload === "object" && payload !== null && "type" in payload && "status" in payload
74 );
75}
76
77export function shouldPublishAgentAwarenessEvent(
78 event: Pick<OrchestrationV2DomainEvent, "type"> & { readonly payload?: unknown },
79): boolean {
80 if (
81 event.type === "thread.created" &&
82 typeof event.payload === "object" &&
83 event.payload !== null &&
84 "historyOrigin" in event.payload &&
85 event.payload.historyOrigin === "v1_import"
86 ) {
87 return false;
88 }
89 // projectThreadAwarenessV2 reads thread metadata, run status, pending requests,
90 // and pending background work (a finished subagent, a cleared roster, or an
91 // ended background item can release a held completion). Message bodies and
92 // tool progress cannot change the published activity.
93 switch (event.type) {
94 case "thread.created":
95 case "thread.archived":
96 case "thread.unarchived":
97 case "thread.deleted":
98 case "thread.metadata-updated":
99 case "thread.pull-request-link-updated":
100 case "thread.pull-request-synced":
101 case "thread.model-selection-updated":
102 case "thread.provider-switched":
103 case "run.created":
104 case "run.updated":
105 case "runtime-request.updated":
106 case "subagent.updated":
107 case "provider-thread.updated":
108 return true;
109 case "thread.settled":
110 case "thread.unsettled":
111 case "thread.snoozed":
112 case "thread.unsnoozed":
113 case "thread.auto-settle-set":
114 case "thread.pinned":
115 case "thread.unpinned":
116 case "thread.pin-reordered":
117 case "thread.active-reordered":
118 case "thread.visited":
119 case "thread.marked-unread":
120 case "thread.runtime-mode-updated":
121 case "thread.interaction-mode-updated":
122 case "run.background-work-cancelled":
123 case "run-attempt.created":
124 case "run-attempt.updated":
125 case "node.updated":
126 case "provider-session.attached":
127 case "provider-session.updated":
128 case "provider-session.detached":
129 case "provider-turn.updated":
130 case "turn-item.updated":
131 return isTurnItemPayload(event.payload) && turnItemUpdateCanEndBackgroundWork(event.payload);
132 case "message.updated":
133 case "plan.updated":
134 case "checkpoint-scope.created":
135 case "checkpoint.captured":
136 case "checkpoint.rollback-requested":
137 case "context-handoff.updated":
138 case "context-transfer.created":
139 case "context-transfer.updated":
140 return false;
141 }
142}
143
144export const makeAgentAwarenessPublishWorker = Effect.fnUntraced(function* <R>(
145 publish: (threadId: ThreadId) => Effect.Effect<void, never, R>,
146) {
147 const queued = new Set<ThreadId>();
148 const worker = yield* makeDrainableWorker((threadId: ThreadId) =>
149 Effect.sync(() => queued.delete(threadId)).pipe(Effect.andThen(publish(threadId))),
150 );
151 const enqueue = (threadId: ThreadId) =>
152 Effect.suspend(() => {
153 if (queued.has(threadId)) return Effect.void;
154 // Removing the ID when processing starts allows one new queue entry
155 // for updates received while the current snapshot is being published.
156 queued.add(threadId);
157 return worker.enqueue(threadId);
158 }).pipe(Effect.uninterruptible);
159 return { enqueue, drain: worker.drain };
160});
161
162function agentAwarenessPublishIdentity(state: RelayAgentActivityState | null): string {
163 if (state === null) {
164 return "null";
165 }
166 const { updatedAt: _updatedAt, ...meaningfulState } = state;
167 return JSON.stringify(meaningfulState);
168}
169
170function resolveAgentActivityPublishingStartupState(input: {
171 readonly relayConfigured: boolean;
172 readonly publishEnabled: boolean;
173}): "waiting-for-link" | "disabled" | "enabled" {
174 if (!input.relayConfigured) {
175 return "waiting-for-link";
176 }
177 return input.publishEnabled ? "enabled" : "disabled";
178}
179
180const RELAY_AGENT_ACTIVITY_DETAIL_MAX_LENGTH = 160;
181const REDACTED_RELAY_AGENT_FAILURE_DETAIL = "The agent run failed.";
182const RELAY_AGENT_ACTIVITY_RETRY_DELAYS_MS = [1_000, 2_000, 4_000, 8_000, 16_000] as const;
183
184function sanitizeRelayAgentActivityState(
185 state: RelayAgentActivityState | null,
186): RelayAgentActivityState | null {
187 if (state === null) {
188 return null;
189 }
190 const { detail: _detail, ...rest } = state;
191 const detail = (state.phase === "failed" ? REDACTED_RELAY_AGENT_FAILURE_DETAIL : state.detail)
192 ?.trim()
193 .slice(0, RELAY_AGENT_ACTIVITY_DETAIL_MAX_LENGTH)
194 .trim();
195 return detail ? { ...rest, detail } : rest;
196}
197
198function relayEnvironmentClient(token: string) {
199 return HttpClient.mapRequest(HttpClientRequest.setHeader("authorization", `Bearer ${token}`));
200}
201
202function deliveryStats(
203 deliveries: ReadonlyArray<{
204 readonly ok: boolean;
205 readonly queued?: boolean | undefined;
206 readonly kind: string;
207 readonly apnsStatus?: number | null;
208 readonly apnsReason?: string | null;
209 }>,
210) {
211 let queued = 0;
212 let successful = 0;
213 let failed = 0;
214 const failedReasons: string[] = [];
215 const kinds = new Set<string>();
216
217 for (const delivery of deliveries) {
218 kinds.add(delivery.kind);
219 if (delivery.queued) {
220 queued += 1;
221 continue;
222 }
223 if (delivery.ok) {
224 successful += 1;
225 continue;
226 }
227 failed += 1;
228 failedReasons.push(`${delivery.apnsStatus ?? "transport"}:${delivery.apnsReason ?? "unknown"}`);
229 }
230
231 return {
232 total: deliveries.length,
233 queued,
234 successful,
235 failed,
236 kinds: [...kinds],
237 failedReasons,
238 };
239}
240
241function signRelayAgentActivityPublishProof(input: {
242 readonly privateKey: string;
243 readonly payload: RelayAgentActivityPublishProofPayload;
244}) {
245 return signRelayJwt({
246 privateKey: input.privateKey,
247 typ: RELAY_ACTIVITY_PUBLISH_TYP,
248 payload: input.payload,
249 });
250}
251
252const makePublishProof = Effect.fn("makePublishProof")(function* (input: {
253 readonly privateKey: string;
254 readonly relayIssuer: string;
255 readonly environmentId: string;
256 readonly threadId: ThreadId;
257 readonly state: RelayAgentActivityState | null;
258 readonly jti: string;
259}) {
260 const now = yield* DateTime.now;
261 const expiresAt = DateTime.add(now, { minutes: 5 });
262 const payload = {
263 iss: `t3-env:${input.environmentId}`,
264 aud: normalizeRelayIssuer(input.relayIssuer),
265 sub: input.environmentId,
266 jti: input.jti,
267 iat: Math.floor(now.epochMilliseconds / 1_000),
268 exp: Math.floor(expiresAt.epochMilliseconds / 1_000),
269 environmentId: input.environmentId as RelayAgentActivityPublishProofPayload["environmentId"],
270 threadId: input.threadId,
271 state: input.state,
272 } satisfies RelayAgentActivityPublishProofPayload;
273 return yield* signRelayAgentActivityPublishProof({ privateKey: input.privateKey, payload });
274});
275
276// Compact, log-safe view of the fields the awareness phase ladder reads.
277function describeThreadShellForAwareness(
278 thread: Option.Option<OrchestrationV2ThreadShell>,
279): Record<string, unknown> {
280 if (Option.isNone(thread)) {
281 return { found: false };
282 }
283 const shell = thread.value;
284 return {
285 found: true,
286 status: shell.status,
287 activityRunStatus: shell.activityRunStatus ?? null,
288 activeRunId: shell.activeRunId ?? null,
289 latestRunId: shell.latestRunId ?? null,
290 pendingRuntimeRequestKind: shell.pendingRuntimeRequest?.kind ?? null,
291 hasActionableProposedPlan: shell.hasActionableProposedPlan,
292 };
293}
294
295function resolveAgentAwarenessRelayPublishSnapshot(input: {
296 readonly environmentId: EnvironmentId;
297 readonly threadId: ThreadId;
298 readonly thread: Option.Option<OrchestrationV2ThreadShell>;
299 readonly project: Option.Option<Project>;
300}): {
301 readonly projectId: string | null;
302 readonly state: RelayAgentActivityState | null;
303 readonly reason: "snapshot" | "thread-not-found" | "project-not-found";
304} {
305 if (Option.isNone(input.thread)) {
306 return {
307 projectId: null,
308 state: null,
309 reason: "thread-not-found",
310 };
311 }
312 if (Option.isNone(input.project)) {
313 return {
314 projectId: input.thread.value.projectId,
315 state: null,
316 reason: "project-not-found",
317 };
318 }
319 return {
320 projectId: input.thread.value.projectId,
321 state: sanitizeRelayAgentActivityState(
322 projectThreadAwarenessV2({
323 environmentId: input.environmentId,
324 project: input.project.value,
325 thread: input.thread.value,
326 }),
327 ),
328 reason: "snapshot",
329 };
330}
331
332function terminalWorkSinceStart(thread: OrchestrationV2ThreadShell, startedAt: number): boolean {
333 return (
334 thread.latestRunCompletedAt != null &&
335 DateTime.toEpochMillis(thread.latestRunCompletedAt) > startedAt
336 );
337}
338
339export function resolveAgentAwarenessRelayActiveThreadIds(input: {
340 readonly environmentId: EnvironmentId;
341 readonly startedAt: number;
342 readonly projects: ReadonlyArray<Pick<Project, "id" | "title">>;
343 readonly threads: ReadonlyArray<OrchestrationV2ThreadShell>;
344}): ReadonlyArray<ThreadId> {
345 const projectById = new Map(input.projects.map((project) => [project.id, project]));
346 return input.threads
347 .filter((thread) => {
348 const project = projectById.get(thread.projectId);
349 if (!project) {
350 return false;
351 }
352 const state = projectThreadAwarenessV2({
353 environmentId: input.environmentId,
354 project,
355 thread,
356 });
357 return (
358 state !== null &&
359 (state.phase !== "completed" && state.phase !== "failed"
360 ? true
361 : terminalWorkSinceStart(thread, input.startedAt))
362 );
363 })
364 .map((thread) => thread.id);
365}
366
367/** @public Service construction is part of the canonical Effect module API. */
368export const make = Effect.gen(function* () {
369 const secrets = yield* ServerSecretStore.ServerSecretStore;
370 const serverEnvironment = yield* ServerEnvironment.ServerEnvironment;
371 const threads = yield* ThreadManagement.ThreadManagementService;
372 const projects = yield* ProjectService.ProjectService;
373 const crypto = yield* Crypto.Crypto;
374 const scope = yield* Effect.scope;
375 const cloudLinkKeyPair = yield* getOrCreateEnvironmentKeyPairFromSecretStore(secrets);
376 const startedAt = (yield* DateTime.now).epochMilliseconds;
377 const activeSnapshotPublishedRef = yield* Ref.make(false);
378 // Holds at most one pending wake, so a burst of requests costs one retry.
379 const catchUpRequests = yield* Queue.dropping<void>(1);
380 const publishedStateByThreadRef = yield* Ref.make(new Map<ThreadId, string>());
381
382 const readSecretString = (name: string) =>
383 secrets
384 .get(name)
385 .pipe(
386 Effect.map((bytes) =>
387 Option.isSome(bytes) ? new TextDecoder().decode(bytes.value) : null,
388 ),
389 );
390
391 const readRelayConfig = Effect.gen(function* () {
392 const [url, issuer, environmentCredential] = yield* Effect.all([
393 readSecretString(RELAY_URL_SECRET),
394 readSecretString(RELAY_ISSUER_SECRET),
395 readSecretString(RELAY_ENVIRONMENT_CREDENTIAL_SECRET),
396 ]);
397 return url && environmentCredential
398 ? { url, issuer: issuer ?? url, environmentCredential }
399 : null;
400 });
401
402 const readPublishAgentActivityEnabled = readSecretString(PUBLISH_AGENT_ACTIVITY_SECRET).pipe(
403 Effect.map(isAgentActivityPublishingEnabledValue),
404 );
405
406 const makeRelayClient = (relayConfig: {
407 readonly url: string;
408 readonly environmentCredential: string;
409 }) =>
410 HttpApiClient.make(RelayApi, {
411 baseUrl: relayConfig.url,
412 transformClient: relayEnvironmentClient(relayConfig.environmentCredential),
413 }).pipe(Effect.provide(FetchHttpClient.layer));
414
415 let cachedRelayClient: Effect.Success<ReturnType<typeof makeRelayClient>> | undefined;
416 let publishedRelayConfig: NonNullable<Effect.Success<typeof readRelayConfig>> | undefined;
417
418 // Deadlines for publishes that need confirmation (tombstones and
419 // first-state completions). The confirming publish is re-enqueued through
420 // the same drainable worker as every other publish, so a confirmed
421 // tombstone can never race an in-flight live update; a recovered state
422 // clears the deadline. Assigned after the worker exists.
423 const publishConfirmDeadlines = new Map<ThreadId, number>();
424 let schedulePublishConfirm: (threadId: ThreadId) => Effect.Effect<void> = () => Effect.void;
425 const publishRetries = new Map<
426 ThreadId,
427 { readonly attempts: number; timer: Fiber.Fiber<void> | undefined }
428 >();
429 const cancelPublishRetry = Effect.fnUntraced(function* (threadId: ThreadId) {
430 const retry = publishRetries.get(threadId);
431 publishRetries.delete(threadId);
432 if (retry?.timer !== undefined) yield* Fiber.interrupt(retry.timer);
433 });
434 const cancelPublishRetries = Effect.suspend(() =>
435 Effect.forEach([...publishRetries.keys()], cancelPublishRetry, { discard: true }),
436 );
437 let schedulePublishRetry: (threadId: ThreadId) => Effect.Effect<void> = () => Effect.void;
438 const resetPublishedConnection = Effect.gen(function* () {
439 publishedRelayConfig = undefined;
440 cachedRelayClient = undefined;
441 publishConfirmDeadlines.clear();
442 yield* Ref.set(publishedStateByThreadRef, new Map());
443 });
444
445 const publishThreadUnsafe = Effect.fn("publishThreadUnsafe")(function* (threadId: ThreadId) {
446 const publishAgentActivity = yield* readPublishAgentActivityEnabled;
447 if (!publishAgentActivity) {
448 yield* cancelPublishRetries;
449 yield* resetPublishedConnection;
450 yield* Effect.logDebug("agent activity publish skipped; publication disabled", {
451 threadId,
452 });
453 return;
454 }
455 const relayConfig = yield* readRelayConfig;
456 if (!relayConfig) {
457 yield* cancelPublishRetries;
458 yield* resetPublishedConnection;
459 yield* Effect.logDebug("agent activity publish skipped; relay link credentials unavailable", {
460 threadId,
461 });
462 return;
463 }
464 if (
465 publishedRelayConfig?.url !== relayConfig.url ||
466 publishedRelayConfig.issuer !== relayConfig.issuer ||
467 publishedRelayConfig.environmentCredential !== relayConfig.environmentCredential
468 ) {
469 yield* resetPublishedConnection;
470 publishedRelayConfig = relayConfig;
471 }
472 const environmentId = yield* serverEnvironment.getEnvironmentId;
473
474 // Only the relay interaction (signing the proof and the call) is exported;
475 // the reads that decide what to publish are the user's local work.
476 const publishState = (input: {
477 readonly projectId: string | null;
478 readonly state: RelayAgentActivityState | null;
479 readonly reason: string;
480 }) =>
481 Effect.gen(function* () {
482 if (cachedRelayClient === undefined) {
483 cachedRelayClient = yield* makeRelayClient(relayConfig);
484 }
485 const relayClient = cachedRelayClient;
486 const proof = yield* makePublishProof({
487 privateKey: cloudLinkKeyPair.privateKey,
488 relayIssuer: relayConfig.issuer,
489 environmentId,
490 threadId,
491 state: input.state,
492 jti: yield* crypto.randomUUIDv4(),
493 });
494
495 yield* Effect.logInfo("publishing agent activity for thread", {
496 environmentId,
497 threadId,
498 projectId: input.projectId,
499 statePhase: input.state?.phase ?? null,
500 hasState: input.state !== null,
501 reason: input.reason,
502 });
503
504 const response = yield* relayClient.server
505 .publishAgentActivity({
506 params: {
507 environmentId,
508 threadId,
509 },
510 payload: {
511 state: input.state,
512 proof,
513 },
514 })
515 .pipe(Effect.timeout("10 seconds"));
516
517 yield* Effect.logInfo("agent activity publish completed", {
518 environmentId,
519 threadId,
520 ok: response.ok,
521 deliveries: deliveryStats(response.deliveries),
522 });
523 }).pipe(Effect.withSpan("relay.agent_activity.publish"));
524
525 // Per-thread shell read: this publish runs for every activity-relevant
526 // domain event, so materializing the full shell here would make the cost
527 // of one thread's activity proportional to how many threads exist.
528 const threadShell = yield* threads.getThreadShell(threadId);
529 if (
530 threadShell?.lineage.relationshipToParent === "subagent" &&
531 !(yield* Ref.get(publishedStateByThreadRef)).has(threadId)
532 ) {
533 // Subagents never project activity, so the relay holds no row to clear.
534 // Their events would otherwise publish a tombstone each, and every
535 // publish re-delivers the user's aggregate. Checked before the archive
536 // filter so archiving one stays quiet too.
537 return;
538 }
539 const thread =
540 threadShell === null || threadShell.archivedAt !== null
541 ? Option.none<OrchestrationV2ThreadShell>()
542 : Option.some(threadShell);
543 const project = Option.isSome(thread)
544 ? yield* projects.getById(thread.value.projectId)
545 : Option.none<Project>();
546 const snapshot = resolveAgentAwarenessRelayPublishSnapshot({
547 environmentId,
548 threadId,
549 thread,
550 project,
551 });
552 const publishIdentity = agentAwarenessPublishIdentity(snapshot.state);
553 const publishedStateByThread = yield* Ref.get(publishedStateByThreadRef);
554 if (
555 (snapshot.state?.phase === "completed" || snapshot.state?.phase === "failed") &&
556 !publishedStateByThread.has(threadId)
557 ) {
558 // Startup has no publish history. Only work from this server process may
559 // produce an initial terminal alert; historical threads remain quiet.
560 if (Option.isNone(thread) || !terminalWorkSinceStart(thread.value, startedAt)) return;
561 }
562 if (publishedStateByThread.get(threadId) === publishIdentity) {
563 // The projection is back at (or never left) the last published state, so
564 // any pending deferred confirmation is moot. Leaving the deadline in
565 // place would let a much later transient null find it already expired
566 // and publish a tombstone immediately, skipping the deferral window.
567 publishConfirmDeadlines.delete(threadId);
568 yield* Effect.logDebug("agent activity publish skipped; projected state unchanged", {
569 environmentId,
570 threadId,
571 reason: snapshot.reason,
572 });
573 return;
574 }
575
576 // Two projections need confirmation before publishing, because both can
577 // appear transiently while the projector is mid-write and publishing them
578 // immediately is destructive or noisy:
579 // - null (tombstone) while the previous published state was live: deletes
580 // the thread from every armed card mid-conversation.
581 // - completed as the thread's FIRST published state: sessions boot at
582 // "ready" before their first turn, which projects as completed for an
583 // instant and sends a spurious Done notification at thread birth.
584 // Defer, schedule a re-publish through the ordinary worker queue, and
585 // only publish if the projection still holds when it drains.
586 const requiresConfirmation =
587 (snapshot.state === null &&
588 publishedStateByThread.get(threadId) !== agentAwarenessPublishIdentity(null)) ||
589 (snapshot.state?.phase === "completed" && !publishedStateByThread.has(threadId));
590 if (requiresConfirmation) {
591 const nowMs = (yield* DateTime.now).epochMilliseconds;
592 const deadline = publishConfirmDeadlines.get(threadId);
593 if (deadline === undefined) {
594 publishConfirmDeadlines.set(threadId, nowMs + 5_000);
595 yield* Effect.logInfo("agent activity publish deferred pending confirmation", {
596 environmentId,
597 threadId,
598 reason: snapshot.reason,
599 statePhase: snapshot.state?.phase ?? null,
600 shell: describeThreadShellForAwareness(thread),
601 });
602 yield* schedulePublishConfirm(threadId);
603 return;
604 }
605 if (nowMs < deadline) {
606 return;
607 }
608 yield* Effect.logInfo("agent activity deferred publish confirmed", {
609 environmentId,
610 threadId,
611 reason: snapshot.reason,
612 statePhase: snapshot.state?.phase ?? null,
613 shell: describeThreadShellForAwareness(thread),
614 });
615 } else {
616 publishConfirmDeadlines.delete(threadId);
617 }
618
619 if (snapshot.reason === "thread-not-found") {
620 yield* Effect.logDebug("publishing agent activity tombstone; thread not found", {
621 environmentId,
622 threadId,
623 });
624 } else if (snapshot.reason === "project-not-found") {
625 yield* Effect.logDebug("publishing agent activity tombstone; project not found", {
626 environmentId,
627 threadId,
628 projectId: snapshot.projectId,
629 });
630 }
631
632 yield* publishState({
633 projectId: snapshot.projectId,
634 state: snapshot.state,
635 reason: snapshot.reason,
636 });
637 publishConfirmDeadlines.delete(threadId);
638 yield* Ref.update(publishedStateByThreadRef, (publishedStates) => {
639 const nextPublishedStates = new Map(publishedStates);
640 if (snapshot.state === null) {
641 nextPublishedStates.delete(threadId);
642 } else {
643 nextPublishedStates.set(threadId, publishIdentity);
644 }
645 return nextPublishedStates;
646 });
647 });
648
649 const processThreadPublish = Effect.fnUntraced(function* (threadId: ThreadId) {
650 const retry = publishRetries.get(threadId);
651 if (retry?.timer !== undefined) {
652 const timer = retry.timer;
653 retry.timer = undefined;
654 yield* Fiber.interrupt(timer);
655 }
656 yield* publishThreadUnsafe(threadId).pipe(
657 Effect.tap(() => cancelPublishRetry(threadId)),
658 Effect.catchCause((cause) => {
659 if (Cause.hasInterruptsOnly(cause)) return Effect.void;
660 return Effect.logWarning("agent activity publish failed", {
661 threadId,
662 cause: Cause.pretty(cause),
663 }).pipe(Effect.andThen(schedulePublishRetry(threadId)));
664 }),
665 Effect.withSpan("AgentAwarenessRelay.publishThread"),
666 );
667 });
668
669 const worker = yield* makeAgentAwarenessPublishWorker(processThreadPublish);
670 const enqueueThreadPublish = (threadId: ThreadId) =>
671 cancelPublishRetry(threadId).pipe(Effect.andThen(worker.enqueue(threadId)));
672 const publishThread: AgentAwarenessRelay["Service"]["publishThread"] = (threadId) =>
673 enqueueThreadPublish(threadId).pipe(Effect.andThen(worker.drain));
674
675 schedulePublishRetry = Effect.fnUntraced(function* (threadId: ThreadId) {
676 const attempts = publishRetries.get(threadId)?.attempts ?? 0;
677 const delayMs = RELAY_AGENT_ACTIVITY_RETRY_DELAYS_MS[attempts];
678 if (delayMs === undefined) {
679 // Keep the exhausted budget until fresh activity or a successful publish
680 // clears it; an old confirmation timer must not start another retry series.
681 yield* Effect.logWarning("agent activity publish retry budget exhausted", { threadId });
682 return;
683 }
684 const retry = {
685 attempts: attempts + 1,
686 timer: undefined as Fiber.Fiber<void> | undefined,
687 };
688 publishRetries.set(threadId, retry);
689 const timer = yield* Effect.sleep(delayMs).pipe(
690 Effect.andThen(
691 Effect.suspend(() => {
692 if (publishRetries.get(threadId) !== retry) return Effect.void;
693 retry.timer = undefined;
694 return worker.enqueue(threadId);
695 }),
696 ),
697 Effect.forkIn(scope),
698 );
699 if (publishRetries.get(threadId) !== retry) {
700 yield* Fiber.interrupt(timer);
701 } else {
702 retry.timer = timer;
703 }
704 });
705
706 // Publishes the active threads once. Returns why it did not, so the retry
707 // knows whether it is waiting on a link or on the publish setting.
708 const publishActiveThreadsUnsafe = Effect.gen(function* () {
709 // One secret read settles the common never-linked case; the full link
710 // config is read only once publishing is on.
711 const relayUrl = yield* readSecretString(RELAY_URL_SECRET).pipe(
712 Effect.orElseSucceed(() => null),
713 );
714 if (!relayUrl) {
715 yield* Effect.logDebug("agent activity snapshot skipped; relay link credentials unavailable");
716 return "unlinked" as const;
717 }
718 const publishAgentActivity = yield* readPublishAgentActivityEnabled.pipe(
719 Effect.orElseSucceed(() => false),
720 );
721 if (!publishAgentActivity) {
722 yield* Effect.logDebug("agent activity snapshot skipped; publication disabled");
723 return "disabled" as const;
724 }
725 const relayConfig = yield* readRelayConfig.pipe(Effect.orElseSucceed(() => null));
726 if (!relayConfig) {
727 yield* Effect.logDebug("agent activity snapshot skipped; relay link credentials unavailable");
728 return "unlinked" as const;
729 }
730 const environmentId = yield* serverEnvironment.getEnvironmentId;
731 const [projectSnapshot, shellSnapshot] = yield* Effect.all([
732 projects.snapshot,
733 threads.getShellSnapshot(),
734 ]);
735 const activeThreadIds = resolveAgentAwarenessRelayActiveThreadIds({
736 environmentId,
737 startedAt,
738 projects: projectSnapshot.projects,
739 threads: shellSnapshot.threads,
740 });
741 if (activeThreadIds.length === 0) {
742 yield* Effect.logDebug("agent activity snapshot has no publishable threads");
743 return "published" as const;
744 }
745 yield* Effect.logInfo("publishing active agent activity snapshot", {
746 count: activeThreadIds.length,
747 });
748 yield* Effect.forEach(activeThreadIds, enqueueThreadPublish, { discard: true });
749 yield* worker.drain;
750 return "published" as const;
751 });
752
753 // Publishes the catch-up snapshot of active threads once the environment is
754 // linked and publishing is enabled. Many environments never link, so while
755 // unlinked the retry backs off from 5 s to 60 s. Only this process writes
756 // the link, and it calls `requestCatchUp`, which ends the wait early. A
757 // linked environment keeps the 5 s retry, because `t3 connect publish` can
758 // turn publishing on from another process.
759 const publishActiveThreadsOnceWhenConfigured = (logEnabledWhenReady: boolean) =>
760 Effect.gen(function* () {
761 let unlinkedRetryDelayMs = 5_000;
762 while (!(yield* Ref.get(activeSnapshotPublishedRef))) {
763 const result = yield* publishActiveThreadsUnsafe.pipe(
764 Effect.orElseSucceed(() => "failed" as const),
765 );
766 if (result === "published") {
767 yield* Ref.set(activeSnapshotPublishedRef, true);
768 if (logEnabledWhenReady) {
769 const relayConfig = yield* readRelayConfig.pipe(Effect.orElseSucceed(() => null));
770 yield* Effect.logInfo("agent activity publishing enabled after link reconciliation", {
771 relayUrl: relayConfig?.url,
772 });
773 }
774 return;
775 }
776 const retryDelayMs = result === "unlinked" ? unlinkedRetryDelayMs : 5_000;
777 yield* Effect.race(Effect.sleep(retryDelayMs), Queue.take(catchUpRequests));
778 if (result === "unlinked") {
779 unlinkedRetryDelayMs = Math.min(unlinkedRetryDelayMs * 2, 60_000);
780 }
781 }
782 });
783
784 schedulePublishConfirm = (threadId) =>
785 Effect.forkIn(
786 Effect.sleep("5 seconds").pipe(
787 Effect.andThen(worker.enqueue(threadId)),
788 Effect.catchCause((cause) =>
789 Effect.logWarning("deferred agent activity confirmation failed", {
790 threadId,
791 cause: Cause.pretty(cause),
792 }),
793 ),
794 ),
795 scope,
796 ).pipe(Effect.asVoid);
797
798 const start: AgentAwarenessRelay["Service"]["start"] = Effect.fn("AgentAwarenessRelay.start")(
799 function* () {
800 const [relayConfig, publishEnabled] = yield* Effect.all([
801 readRelayConfig.pipe(Effect.orElseSucceed(() => null)),
802 readPublishAgentActivityEnabled.pipe(Effect.orElseSucceed(() => false)),
803 ]);
804 const startupState = resolveAgentActivityPublishingStartupState({
805 relayConfigured: relayConfig !== null,
806 publishEnabled,
807 });
808 switch (startupState) {
809 case "waiting-for-link":
810 yield* Effect.logInfo(
811 "agent activity publishing standby; waiting for T3 Connect link reconciliation",
812 );
813 break;
814 case "disabled":
815 yield* Effect.logInfo("agent activity publishing disabled by T3 Connect configuration");
816 break;
817 case "enabled":
818 yield* Effect.logInfo("agent activity publishing enabled", {
819 relayUrl: relayConfig?.url,
820 });
821 break;
822 }
823 yield* forkParked(
824 Effect.sleep("1 second").pipe(
825 Effect.andThen(publishActiveThreadsOnceWhenConfigured(startupState !== "enabled")),
826 ),
827 );
828 yield* forkParked(
829 Stream.runForEach(threads.streamDomainEvents, (event) => {
830 const threadId = eventThreadId(event);
831 if (!shouldPublishAgentAwarenessEvent(event)) {
832 return Effect.void;
833 }
834 return Effect.logDebug("agent activity publishing queued thread publish", {
835 eventType: event.type,
836 threadId,
837 }).pipe(Effect.andThen(enqueueThreadPublish(threadId)));
838 }),
839 );
840 },
841 );
842
843 return AgentAwarenessRelay.of({
844 publishThread,
845 drain: worker.drain,
846 requestCatchUp: () => Queue.offer(catchUpRequests, undefined).pipe(Effect.asVoid),
847 start,
848 });
849});
850
851export const layer = Layer.effect(AgentAwarenessRelay, make);