1import type { AgentAwarenessThread } from "../orchestration-v2/ProjectionStore.ts";
2import {
3 isListedThreadLineage,
4 type EnvironmentId,
5 type OrchestrationV2DomainEvent,
6 type OrchestrationV2TurnItem,
7 type Project,
8 type RuntimeRequestId,
9 type ThreadId,
10} from "@t3tools/contracts";
11import {
12 RelayApi,
13 type RelayAgentActivityPublishProofPayload,
14 type RelayAgentActivityState,
15} from "@t3tools/contracts/relay";
16import {
17 type SubagentRequestAwareness,
18 projectThreadAwarenessV2,
19 subagentAwaitsUser,
20 turnItemCanChangeUserWait,
21} from "@t3tools/shared/agentAwareness";
22import { turnItemUpdateCanEndBackgroundWork } from "@t3tools/shared/orchestrationV2PendingBackgroundWork";
23import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker";
24import { withRelayClientTracing } from "@t3tools/shared/relayTracing";
25import { normalizeRelayIssuer, RELAY_ACTIVITY_PUBLISH_TYP } from "@t3tools/shared/relayJwt";
26import * as Cause from "effect/Cause";
27import * as Context from "effect/Context";
28import * as SecureRandom from "effect/crypto/SecureRandom";
29import * as DateTime from "effect/DateTime";
30import * as Effect from "effect/Effect";
31import * as Fiber from "effect/Fiber";
32import * as Layer from "effect/Layer";
33import * as Option from "effect/Option";
34import * as Queue from "effect/Queue";
35import * as Ref from "effect/Ref";
36import type * as Scope from "effect/Scope";
37import * as Stream from "effect/Stream";
38import * as FetchHttpClient from "effect/http/FetchHttpClient";
39import * as HttpClient from "effect/http/HttpClient";
40import * as HttpClientRequest from "effect/http/HttpClientRequest";
41import * as HttpApiClient from "effect/http-api/HttpApiClient";
42
43import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
44import {
45 isAgentActivityPublishingEnabledValue,
46 PUBLISH_AGENT_ACTIVITY_SECRET,
47 RELAY_ENVIRONMENT_CREDENTIAL_SECRET,
48 RELAY_ISSUER_SECRET,
49 RELAY_URL_SECRET,
50} from "../cloud/config.ts";
51import { EnvironmentSigningKey } from "../cloud/environmentKeys.ts";
52import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
53import * as ThreadManagement from "../orchestration-v2/ThreadManagementService.ts";
54import * as ProjectService from "../project/ProjectService.ts";
55import { forkParked } from "../serverActivation.ts";
56
57export class AgentAwarenessRelay extends Context.Service<
58 AgentAwarenessRelay,
59 {
60 readonly publishThread: (threadId: ThreadId) => Effect.Effect<void>;
61 readonly drain: Effect.Effect<void>;
62 readonly requestCatchUp: () => Effect.Effect<void>;
63 readonly start: () => Effect.Effect<void, never, Scope.Scope>;
64 }
65>()("t3/relay/AgentAwarenessRelay") {}
66
67function eventThreadId(event: OrchestrationV2DomainEvent): ThreadId {
68 return event.threadId;
69}
70
71function isTurnItemPayload(
72 payload: unknown,
73): payload is Pick<OrchestrationV2TurnItem, "type" | "status"> {
74 return (
75 typeof payload === "object" && payload !== null && "type" in payload && "status" in payload
76 );
77}
78
79function shouldPublishAgentAwarenessEvent(
80 event: Pick<OrchestrationV2DomainEvent, "type"> & { readonly payload?: unknown },
81): boolean {
82 if (
83 event.type === "thread.created" &&
84 typeof event.payload === "object" &&
85 event.payload !== null &&
86 "historyOrigin" in event.payload &&
87 event.payload.historyOrigin === "v1_import"
88 ) {
89 return false;
90 }
91 switch (event.type) {
92 case "thread.created":
93 case "thread.archived":
94 case "thread.unarchived":
95 case "thread.deleted":
96 case "thread.metadata-updated":
97 case "thread.pull-request-link-updated":
98 case "thread.pull-request-synced":
99 case "thread.settled":
100 case "thread.model-selection-updated":
101 case "thread.provider-switched":
102 case "run.created":
103 case "run.updated":
104 case "runtime-request.updated":
105 case "subagent.updated":
106 case "provider-thread.updated":
107 return true;
108 // Unsettling must not re-alert a dismissed result.
109 case "thread.unsettled":
110 case "thread.guest-share-updated":
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.tags-updated":
119 case "thread.visited":
120 case "thread.marked-unread":
121 case "thread.runtime-mode-updated":
122 case "thread.interaction-mode-updated":
123 case "run.background-work-cancelled":
124 case "run-attempt.created":
125 case "run-attempt.updated":
126 case "node.updated":
127 case "provider-session.attached":
128 case "provider-session.updated":
129 case "provider-session.detached":
130 case "provider-turn.updated":
131 case "turn-item.updated":
132 return (
133 isTurnItemPayload(event.payload) &&
134 (turnItemUpdateCanEndBackgroundWork(event.payload) ||
135 turnItemCanChangeUserWait(event.payload))
136 );
137 case "message.updated":
138 case "plan.updated":
139 case "checkpoint-scope.created":
140 case "checkpoint.captured":
141 case "checkpoint.rollback-requested":
142 case "context-handoff.updated":
143 case "context-transfer.created":
144 case "context-transfer.updated":
145 return false;
146 }
147}
148
149export const makeAgentAwarenessPublishWorker = Effect.fnUntraced(function* <R>(
150 publish: (threadId: ThreadId) => Effect.Effect<void, never, R>,
151) {
152 const queued = new Set<ThreadId>();
153 const worker = yield* makeDrainableWorker((threadId: ThreadId) =>
154 Effect.sync(() => queued.delete(threadId)).pipe(Effect.andThen(publish(threadId))),
155 );
156 const enqueue = (threadId: ThreadId) =>
157 Effect.suspend(() => {
158 if (queued.has(threadId)) return Effect.void;
159 // Remove queued IDs when processing starts so publication can queue one later update.
160 queued.add(threadId);
161 return worker.enqueue(threadId);
162 }).pipe(Effect.uninterruptible);
163 return { enqueue, drain: worker.drain };
164});
165
166function agentAwarenessPublishIdentity(
167 state: RelayAgentActivityState | null,
168 subagentRequest: SubagentRequestAwareness | null = null,
169): string {
170 if (state === null) {
171 return "null";
172 }
173 const { updatedAt: _updatedAt, ...meaningfulState } = state;
174 // A child's next request reads the same as its last one, but it is a new
175 // wait and must reach the phone.
176 return JSON.stringify(
177 subagentRequest === null ? meaningfulState : [meaningfulState, subagentRequest.requestId],
178 );
179}
180
181const CLEARED_PUBLISH_IDENTITY = agentAwarenessPublishIdentity(null);
182
183// A published tombstone stays recorded so repeats are skipped, but it is not live state.
184function hasLivePublishedState(
185 publishedStateByThread: ReadonlyMap<ThreadId, string>,
186 threadId: ThreadId,
187): boolean {
188 const identity = publishedStateByThread.get(threadId);
189 return identity !== undefined && identity !== CLEARED_PUBLISH_IDENTITY;
190}
191
192function resolveAgentActivityPublishingStartupState(input: {
193 readonly relayConfigured: boolean;
194 readonly publishEnabled: boolean;
195}): "waiting-for-link" | "disabled" | "enabled" {
196 if (!input.relayConfigured) {
197 return "waiting-for-link";
198 }
199 return input.publishEnabled ? "enabled" : "disabled";
200}
201
202const RELAY_AGENT_ACTIVITY_DETAIL_MAX_LENGTH = 160;
203const REDACTED_RELAY_AGENT_FAILURE_DETAIL = "The agent run failed.";
204const RELAY_AGENT_ACTIVITY_RETRY_DELAYS_MS = [1_000, 2_000, 4_000, 8_000, 16_000] as const;
205
206function sanitizeRelayAgentActivityState(
207 state: RelayAgentActivityState | null,
208): RelayAgentActivityState | null {
209 if (state === null) {
210 return null;
211 }
212 const { detail: _detail, ...rest } = state;
213 const detail = (state.phase === "failed" ? REDACTED_RELAY_AGENT_FAILURE_DETAIL : state.detail)
214 ?.trim()
215 .slice(0, RELAY_AGENT_ACTIVITY_DETAIL_MAX_LENGTH)
216 .trim();
217 return detail ? { ...rest, detail } : rest;
218}
219
220function relayEnvironmentClient(token: string) {
221 return HttpClient.mapRequest(HttpClientRequest.setHeader("authorization", `Bearer ${token}`));
222}
223
224function deliveryStats(
225 deliveries: ReadonlyArray<{
226 readonly ok: boolean;
227 readonly queued?: boolean | undefined;
228 readonly kind: string;
229 readonly apnsStatus?: number | null;
230 readonly apnsReason?: string | null;
231 }>,
232) {
233 let queued = 0;
234 let successful = 0;
235 let failed = 0;
236 const failedReasons: string[] = [];
237 const kinds = new Set<string>();
238
239 for (const delivery of deliveries) {
240 kinds.add(delivery.kind);
241 if (delivery.queued) {
242 queued += 1;
243 continue;
244 }
245 if (delivery.ok) {
246 successful += 1;
247 continue;
248 }
249 failed += 1;
250 failedReasons.push(`${delivery.apnsStatus ?? "transport"}:${delivery.apnsReason ?? "unknown"}`);
251 }
252
253 return {
254 total: deliveries.length,
255 queued,
256 successful,
257 failed,
258 kinds: [...kinds],
259 failedReasons,
260 };
261}
262
263const makePublishProof = Effect.fn("makePublishProof")(function* (input: {
264 readonly relayIssuer: string;
265 readonly environmentId: string;
266 readonly threadId: ThreadId;
267 readonly state: RelayAgentActivityState | null;
268 readonly jti: string;
269}) {
270 const now = yield* DateTime.now;
271 const expiresAt = DateTime.add(now, { minutes: 5 });
272 const payload = {
273 iss: `t3-env:${input.environmentId}`,
274 aud: normalizeRelayIssuer(input.relayIssuer),
275 sub: input.environmentId,
276 jti: input.jti,
277 iat: Math.floor(now.epochMilliseconds / 1_000),
278 exp: Math.floor(expiresAt.epochMilliseconds / 1_000),
279 environmentId: input.environmentId as RelayAgentActivityPublishProofPayload["environmentId"],
280 threadId: input.threadId,
281 state: input.state,
282 } satisfies RelayAgentActivityPublishProofPayload;
283 const signingKey = yield* EnvironmentSigningKey;
284 return yield* signingKey.sign({ typ: RELAY_ACTIVITY_PUBLISH_TYP, payload });
285});
286
287function describeThreadShellForAwareness(
288 thread: Option.Option<AgentAwarenessThread>,
289): Record<string, unknown> {
290 if (Option.isNone(thread)) {
291 return { found: false };
292 }
293 const shell = thread.value;
294 return {
295 found: true,
296 status: shell.status,
297 activityRunStatus: shell.activityRunStatus ?? null,
298 activeRunId: shell.activeRunId ?? null,
299 latestRunId: shell.latestRunId ?? null,
300 pendingRuntimeRequestKind: shell.pendingRuntimeRequest?.kind ?? null,
301 hasActionableProposedPlan: shell.hasActionableProposedPlan,
302 };
303}
304
305interface SubagentRequest {
306 readonly topLevelThreadId: ThreadId;
307 readonly requestId: RuntimeRequestId;
308}
309
310interface SubagentAlert {
311 readonly requestId: RuntimeRequestId;
312 readonly updatedAt: string;
313}
314
315const waitingApprovalId = (thread: AgentAwarenessThread | null) =>
316 thread !== null && thread.archivedAt === null && subagentAwaitsUser(thread)
317 ? thread.pendingRuntimeRequest?.id
318 : undefined;
319
320function resolveAgentAwarenessRelayPublishSnapshot(input: {
321 readonly environmentId: EnvironmentId;
322 readonly thread: Option.Option<AgentAwarenessThread>;
323 readonly project: Option.Option<Project>;
324 readonly pendingSubagentRequest: SubagentRequestAwareness | null;
325}): {
326 readonly projectId: string | null;
327 readonly state: RelayAgentActivityState | null;
328 /** The child request behind the state's wait, when the parent isn't waiting itself. */
329 readonly subagentWait: SubagentRequestAwareness | null;
330 readonly reason: "snapshot" | "thread-not-found" | "project-not-found";
331} {
332 if (Option.isNone(input.thread)) {
333 return {
334 projectId: null,
335 state: null,
336 subagentWait: null,
337 reason: "thread-not-found",
338 };
339 }
340 if (Option.isNone(input.project)) {
341 return {
342 projectId: input.thread.value.projectId,
343 state: null,
344 subagentWait: null,
345 reason: "project-not-found",
346 };
347 }
348 const awarenessInput = {
349 environmentId: input.environmentId,
350 project: input.project.value,
351 thread: input.thread.value,
352 };
353 const state = projectThreadAwarenessV2({
354 ...awarenessInput,
355 pendingSubagentRequest: input.pendingSubagentRequest,
356 });
357 const childDrivesWait =
358 input.pendingSubagentRequest !== null &&
359 state?.phase !== projectThreadAwarenessV2(awarenessInput)?.phase;
360 return {
361 projectId: input.thread.value.projectId,
362 state: sanitizeRelayAgentActivityState(state),
363 subagentWait: childDrivesWait ? input.pendingSubagentRequest : null,
364 reason: "snapshot",
365 };
366}
367
368function terminalWorkSinceStart(thread: AgentAwarenessThread, startedAt: number): boolean {
369 return (
370 thread.latestRunCompletedAt != null &&
371 DateTime.toEpochMillis(thread.latestRunCompletedAt) > startedAt
372 );
373}
374
375export function resolveAgentAwarenessRelayActiveThreadIds(input: {
376 readonly environmentId: EnvironmentId;
377 readonly startedAt: number;
378 readonly projects: ReadonlyArray<Pick<Project, "id" | "title">>;
379 readonly threads: ReadonlyArray<AgentAwarenessThread>;
380}): ReadonlyArray<ThreadId> {
381 const projectById = new Map(input.projects.map((project) => [project.id, project]));
382 return input.threads
383 .filter((thread) => {
384 // Publishing a waiting subagent republishes its top-level thread.
385 if (subagentAwaitsUser(thread)) return true;
386 const project = projectById.get(thread.projectId);
387 if (!project) {
388 return false;
389 }
390 const state = projectThreadAwarenessV2({
391 environmentId: input.environmentId,
392 project,
393 thread,
394 });
395 return (
396 state !== null &&
397 (state.phase !== "completed" && state.phase !== "failed"
398 ? true
399 : terminalWorkSinceStart(thread, input.startedAt))
400 );
401 })
402 .map((thread) => thread.id);
403}
404
405/** @public */
406export const make = Effect.gen(function* () {
407 const secrets = yield* ServerSecretStore.ServerSecretStore;
408 const serverEnvironment = yield* ServerEnvironment.ServerEnvironment;
409 const threads = yield* ThreadManagement.ThreadManagementService;
410 const projects = yield* ProjectService.ProjectService;
411 const random = yield* SecureRandom.SecureRandom;
412 const signingKey = yield* EnvironmentSigningKey;
413 const scope = yield* Effect.scope;
414 const startedAt = (yield* DateTime.now).epochMilliseconds;
415 const catchUpRequests = yield* Queue.dropping<void>(1);
416 const publishedStateByThreadRef = yield* Ref.make(new Map<ThreadId, string>());
417 // Android keys alerts by phase and time, so a child's unchanged request keeps
418 // its first alert time through parent renames, relinks and the parent's own requests.
419 const childAlertByThread = new Map<ThreadId, SubagentAlert>();
420 const subagentRequests = new Map<ThreadId, SubagentRequest>();
421
422 const readSecretString = (name: string) =>
423 secrets
424 .get(name)
425 .pipe(
426 Effect.map((bytes) =>
427 Option.isSome(bytes) ? new TextDecoder().decode(bytes.value) : null,
428 ),
429 );
430
431 const readRelayConfig = Effect.gen(function* () {
432 const [url, issuer, environmentCredential] = yield* Effect.all([
433 readSecretString(RELAY_URL_SECRET),
434 readSecretString(RELAY_ISSUER_SECRET),
435 readSecretString(RELAY_ENVIRONMENT_CREDENTIAL_SECRET),
436 ]);
437 return url && environmentCredential
438 ? { url, issuer: issuer ?? url, environmentCredential }
439 : null;
440 });
441
442 const readPublishAgentActivityEnabled = readSecretString(PUBLISH_AGENT_ACTIVITY_SECRET).pipe(
443 Effect.map(isAgentActivityPublishingEnabledValue),
444 );
445
446 const makeRelayClient = (relayConfig: {
447 readonly url: string;
448 readonly environmentCredential: string;
449 }) =>
450 HttpApiClient.make(RelayApi, {
451 baseUrl: relayConfig.url,
452 transformClient: relayEnvironmentClient(relayConfig.environmentCredential),
453 }).pipe(Effect.provide(FetchHttpClient.layer));
454
455 let cachedRelayClient: Effect.Success<ReturnType<typeof makeRelayClient>> | undefined;
456 let publishedRelayConfig: NonNullable<Effect.Success<typeof readRelayConfig>> | undefined;
457
458 // Confirm through the same worker so tombstones cannot race live updates; recovery clears the deadline.
459 const publishConfirmDeadlines = new Map<ThreadId, number>();
460 // Threads whose confirmation holds a completion, so a thread that settles during the hold still alerts.
461 const heldCompletions = new Set<ThreadId>();
462 let schedulePublishConfirm: (threadId: ThreadId) => Effect.Effect<void> = () => Effect.void;
463 const publishRetries = new Map<
464 ThreadId,
465 { readonly attempts: number; timer: Fiber.Fiber<void> | undefined }
466 >();
467 const cancelPublishRetry = Effect.fnUntraced(function* (threadId: ThreadId) {
468 const retry = publishRetries.get(threadId);
469 publishRetries.delete(threadId);
470 if (retry?.timer !== undefined) yield* Fiber.interrupt(retry.timer);
471 });
472 const cancelPublishRetries = Effect.suspend(() =>
473 Effect.forEach([...publishRetries.keys()], cancelPublishRetry, { discard: true }),
474 );
475 let schedulePublishRetry: (threadId: ThreadId) => Effect.Effect<void> = () => Effect.void;
476 let enqueueTopLevelPublish: (threadId: ThreadId) => Effect.Effect<void> = () => Effect.void;
477 const resetPublishedConnection = Effect.gen(function* () {
478 publishedRelayConfig = undefined;
479 cachedRelayClient = undefined;
480 publishConfirmDeadlines.clear();
481 heldCompletions.clear();
482 yield* Ref.set(publishedStateByThreadRef, new Map());
483 });
484
485 const publishThreadUnsafe = Effect.fn("publishThreadUnsafe")(function* (threadId: ThreadId) {
486 const publishAgentActivity = yield* readPublishAgentActivityEnabled;
487 if (!publishAgentActivity) {
488 yield* cancelPublishRetries;
489 yield* resetPublishedConnection;
490 yield* Effect.logDebug("agent activity publish skipped; publication disabled", {
491 threadId,
492 });
493 return;
494 }
495 const relayConfig = yield* readRelayConfig;
496 if (!relayConfig) {
497 yield* cancelPublishRetries;
498 yield* resetPublishedConnection;
499 yield* Effect.logDebug("agent activity publish skipped; relay link credentials unavailable", {
500 threadId,
501 });
502 return;
503 }
504 if (
505 publishedRelayConfig?.url !== relayConfig.url ||
506 publishedRelayConfig.issuer !== relayConfig.issuer ||
507 publishedRelayConfig.environmentCredential !== relayConfig.environmentCredential
508 ) {
509 yield* resetPublishedConnection;
510 publishedRelayConfig = relayConfig;
511 }
512 const environmentId = yield* serverEnvironment.getEnvironmentId;
513
514 const publishState = Effect.fn("relay.agent_activity.publish")(function* (input: {
515 readonly projectId: string | null;
516 readonly state: RelayAgentActivityState | null;
517 readonly reason: string;
518 }) {
519 if (cachedRelayClient === undefined) {
520 cachedRelayClient = yield* makeRelayClient(relayConfig);
521 }
522 const relayClient = cachedRelayClient;
523 const proof = yield* makePublishProof({
524 relayIssuer: relayConfig.issuer,
525 environmentId,
526 threadId,
527 state: input.state,
528 jti: yield* random.uuidV4(),
529 }).pipe(Effect.provideService(EnvironmentSigningKey, signingKey));
530
531 yield* Effect.logInfo("publishing agent activity for thread", {
532 environmentId,
533 threadId,
534 projectId: input.projectId,
535 statePhase: input.state?.phase ?? null,
536 hasState: input.state !== null,
537 reason: input.reason,
538 });
539
540 const response = yield* relayClient.server
541 .publishAgentActivity({
542 params: {
543 environmentId,
544 threadId,
545 },
546 payload: {
547 state: input.state,
548 proof,
549 },
550 })
551 .pipe(Effect.timeout("10 seconds"));
552
553 yield* Effect.logInfo("agent activity publish completed", {
554 environmentId,
555 threadId,
556 ok: response.ok,
557 deliveries: deliveryStats(response.deliveries),
558 });
559 });
560
561 // Read only this thread; full shell reads would make each activity event scale with all threads.
562 const threadShell = yield* readAwarenessThread(threadId);
563 yield* trackSubagentRequest(threadId, threadShell);
564 if (
565 threadShell !== null &&
566 !isListedThreadLineage(threadShell.lineage) &&
567 !hasLivePublishedState(yield* Ref.get(publishedStateByThreadRef), threadId)
568 ) {
569 // Subagents and side chats have no relay rows to clear; per-event tombstones would needlessly redeliver user aggregates.
570 return;
571 }
572 const thread =
573 threadShell === null || threadShell.archivedAt !== null
574 ? Option.none<AgentAwarenessThread>()
575 : Option.some(threadShell);
576 const project = Option.isSome(thread)
577 ? yield* projects.getById(thread.value.projectId)
578 : Option.none<Project>();
579 const snapshot = resolveAgentAwarenessRelayPublishSnapshot({
580 environmentId,
581 thread,
582 project,
583 pendingSubagentRequest: yield* currentSubagentRequestFor(threadId),
584 });
585 const publishIdentity = agentAwarenessPublishIdentity(snapshot.state, snapshot.subagentWait);
586 const publishedStateByThread = yield* Ref.get(publishedStateByThreadRef);
587 if (
588 (snapshot.state?.phase === "completed" || snapshot.state?.phase === "failed") &&
589 !hasLivePublishedState(publishedStateByThread, threadId)
590 ) {
591 if (Option.isNone(thread) || !terminalWorkSinceStart(thread.value, startedAt)) return;
592 }
593 if (
594 publishedStateByThread.get(threadId) === publishIdentity &&
595 // A held completion settled after an earlier clear still has to alert.
596 !(snapshot.state === null && heldCompletions.has(threadId))
597 ) {
598 // Clear moot deadlines so later transient nulls cannot skip their deferral window.
599 publishConfirmDeadlines.delete(threadId);
600 heldCompletions.delete(threadId);
601 yield* Effect.logDebug("agent activity publish skipped; projected state unchanged", {
602 environmentId,
603 threadId,
604 reason: snapshot.reason,
605 });
606 return;
607 }
608
609 // Confirm tombstones over live state to avoid removing active cards.
610 // Confirm completion: a run can end moments before its continuation starts, and the relay
611 // alerts on every move into completed.
612 const requiresConfirmation = snapshot.state === null || snapshot.state?.phase === "completed";
613 if (requiresConfirmation) {
614 if (snapshot.state?.phase === "completed") heldCompletions.add(threadId);
615 const nowMs = (yield* DateTime.now).epochMilliseconds;
616 const deadline = publishConfirmDeadlines.get(threadId);
617 if (deadline === undefined) {
618 publishConfirmDeadlines.set(threadId, nowMs + 5_000);
619 yield* Effect.logInfo("agent activity publish deferred pending confirmation", {
620 environmentId,
621 threadId,
622 reason: snapshot.reason,
623 statePhase: snapshot.state?.phase ?? null,
624 shell: describeThreadShellForAwareness(thread),
625 });
626 yield* schedulePublishConfirm(threadId);
627 return;
628 }
629 if (nowMs < deadline) {
630 return;
631 }
632 yield* Effect.logInfo("agent activity deferred publish confirmed", {
633 environmentId,
634 threadId,
635 reason: snapshot.reason,
636 statePhase: snapshot.state?.phase ?? null,
637 shell: describeThreadShellForAwareness(thread),
638 });
639 // Settled during the hold, not resumed: alert the current completion, then clear the card
640 // on a new window. Only settlement may hide it; queued or silent runs also project to null.
641 const unsettled =
642 snapshot.state === null &&
643 heldCompletions.has(threadId) &&
644 Option.isSome(thread) &&
645 thread.value.settledOverride === "settled"
646 ? resolveAgentAwarenessRelayPublishSnapshot({
647 environmentId,
648 thread: Option.some({ ...thread.value, settledOverride: null }),
649 project,
650 pendingSubagentRequest: null,
651 })
652 : null;
653 if (unsettled?.state?.phase === "completed") {
654 const completion = unsettled.state;
655 yield* publishState({
656 projectId: unsettled.projectId,
657 state: completion,
658 reason: unsettled.reason,
659 }).pipe(withRelayClientTracing);
660 heldCompletions.delete(threadId);
661 publishConfirmDeadlines.set(threadId, nowMs + 5_000);
662 yield* Ref.update(publishedStateByThreadRef, (publishedStates) =>
663 new Map(publishedStates).set(threadId, agentAwarenessPublishIdentity(completion)),
664 );
665 yield* schedulePublishConfirm(threadId);
666 return;
667 }
668 } else {
669 publishConfirmDeadlines.delete(threadId);
670 heldCompletions.delete(threadId);
671 }
672
673 if (snapshot.reason === "thread-not-found") {
674 yield* Effect.logDebug("publishing agent activity tombstone; thread not found", {
675 environmentId,
676 threadId,
677 });
678 } else if (snapshot.reason === "project-not-found") {
679 yield* Effect.logDebug("publishing agent activity tombstone; project not found", {
680 environmentId,
681 threadId,
682 projectId: snapshot.projectId,
683 });
684 }
685
686 const childRequestId = snapshot.subagentWait?.requestId;
687 const childAlert = childAlertByThread.get(threadId);
688 const state =
689 snapshot.state !== null &&
690 childRequestId !== undefined &&
691 childAlert?.requestId === childRequestId
692 ? { ...snapshot.state, updatedAt: childAlert.updatedAt }
693 : snapshot.state;
694 yield* publishState({
695 projectId: snapshot.projectId,
696 state,
697 reason: snapshot.reason,
698 }).pipe(withRelayClientTracing);
699 if (state !== null && childRequestId !== undefined) {
700 childAlertByThread.set(threadId, { requestId: childRequestId, updatedAt: state.updatedAt });
701 }
702 publishConfirmDeadlines.delete(threadId);
703 heldCompletions.delete(threadId);
704 yield* Ref.update(publishedStateByThreadRef, (publishedStates) =>
705 new Map(publishedStates).set(threadId, publishIdentity),
706 );
707 });
708
709 const readAwarenessThread = (threadId: ThreadId) =>
710 threads
711 .getAgentAwarenessSnapshot(threadId)
712 .pipe(Effect.map((snapshot) => snapshot.threads[0] ?? null));
713
714 // The subagent helpers below are untraced: they run per event inside
715 // publishThreadUnsafe, whose span already covers them.
716 const findTopLevelThreadId = Effect.fnUntraced(function* (child: AgentAwarenessThread) {
717 const seen = new Set<ThreadId>([child.id]);
718 let current: AgentAwarenessThread | null = child;
719 while (
720 current?.lineage.relationshipToParent === "subagent" ||
721 current?.lineage.sideChat === true
722 ) {
723 const parentId: ThreadId | null = current.lineage.parentThreadId;
724 if (parentId === null || seen.has(parentId)) return null;
725 seen.add(parentId);
726 current = yield* readAwarenessThread(parentId);
727 }
728 return current?.id ?? null;
729 });
730
731 const trackSubagentRequest = Effect.fnUntraced(function* (
732 threadId: ThreadId,
733 threadShell: AgentAwarenessThread | null,
734 ) {
735 const requestId = waitingApprovalId(threadShell);
736 const previous = subagentRequests.get(threadId);
737 // An unchanged wait still republishes its parent: a relink catch-up reaches it only here.
738 if (threadShell === null || requestId === undefined) {
739 if (previous === undefined) return;
740 subagentRequests.delete(threadId);
741 yield* enqueueTopLevelPublish(previous.topLevelThreadId);
742 return;
743 }
744 const topLevelThreadId =
745 previous?.topLevelThreadId ?? (yield* findTopLevelThreadId(threadShell));
746 if (topLevelThreadId === null) return;
747 subagentRequests.set(threadId, { topLevelThreadId, requestId });
748 yield* enqueueTopLevelPublish(topLevelThreadId);
749 });
750
751 // The map can miss a child's update while publishing is off or the relay is
752 // unlinked, so each waiting child is re-read before its parent is published.
753 const currentSubagentRequestFor = Effect.fnUntraced(function* (topLevelThreadId: ThreadId) {
754 let pending: SubagentRequestAwareness | null = null;
755 for (const [childId, entry] of subagentRequests) {
756 if (entry.topLevelThreadId !== topLevelThreadId) continue;
757 const child = yield* readAwarenessThread(childId);
758 const requestId = waitingApprovalId(child);
759 if (child === null || requestId === undefined) {
760 subagentRequests.delete(childId);
761 continue;
762 }
763 if (requestId !== entry.requestId) subagentRequests.set(childId, { ...entry, requestId });
764 // The newest wait is the one to announce.
765 if (pending === null || DateTime.isGreaterThan(child.updatedAt, pending.updatedAt)) {
766 pending = {
767 requestId,
768 updatedAt: child.updatedAt,
769 waitsForInput: child.pendingRuntimeRequest?.kind === "user_input",
770 };
771 }
772 }
773 return pending;
774 });
775
776 const processThreadPublish = Effect.fnUntraced(function* (threadId: ThreadId) {
777 const retry = publishRetries.get(threadId);
778 if (retry?.timer !== undefined) {
779 const timer = retry.timer;
780 retry.timer = undefined;
781 yield* Fiber.interrupt(timer);
782 }
783 yield* publishThreadUnsafe(threadId).pipe(
784 Effect.tap(() => cancelPublishRetry(threadId)),
785 Effect.catchCause((cause) => {
786 if (Cause.hasInterruptsOnly(cause)) return Effect.void;
787 return Effect.logWarning("agent activity publish failed", {
788 threadId,
789 cause: Cause.pretty(cause),
790 }).pipe(Effect.andThen(schedulePublishRetry(threadId)));
791 }),
792 Effect.withSpan("AgentAwarenessRelay.publishThread"),
793 );
794 });
795
796 const worker = yield* makeAgentAwarenessPublishWorker(processThreadPublish);
797 const enqueueThreadPublish = (threadId: ThreadId) =>
798 cancelPublishRetry(threadId).pipe(Effect.andThen(worker.enqueue(threadId)));
799 enqueueTopLevelPublish = enqueueThreadPublish;
800 const publishThread: AgentAwarenessRelay["Service"]["publishThread"] = (threadId) =>
801 enqueueThreadPublish(threadId).pipe(Effect.andThen(worker.drain));
802
803 schedulePublishRetry = Effect.fnUntraced(function* (threadId: ThreadId) {
804 const attempts = publishRetries.get(threadId)?.attempts ?? 0;
805 const delayMs = RELAY_AGENT_ACTIVITY_RETRY_DELAYS_MS[attempts];
806 if (delayMs === undefined) {
807 // Retain exhausted budgets until activity or success resets them; old timers must not restart retries.
808 yield* Effect.logWarning("agent activity publish retry budget exhausted", { threadId });
809 return;
810 }
811 const retry = {
812 attempts: attempts + 1,
813 timer: undefined as Fiber.Fiber<void> | undefined,
814 };
815 publishRetries.set(threadId, retry);
816 const timer = yield* Effect.sleep(delayMs).pipe(
817 Effect.andThen(
818 Effect.suspend(() => {
819 if (publishRetries.get(threadId) !== retry) return Effect.void;
820 retry.timer = undefined;
821 return worker.enqueue(threadId);
822 }),
823 ),
824 Effect.forkIn(scope),
825 );
826 if (publishRetries.get(threadId) !== retry) {
827 yield* Fiber.interrupt(timer);
828 } else {
829 retry.timer = timer;
830 }
831 });
832
833 const publishActiveThreadsUnsafe = Effect.gen(function* () {
834 const relayUrl = yield* readSecretString(RELAY_URL_SECRET).pipe(
835 Effect.orElseSucceed(() => null),
836 );
837 if (!relayUrl) {
838 yield* Effect.logDebug("agent activity snapshot skipped; relay link credentials unavailable");
839 return "unlinked" as const;
840 }
841 const publishAgentActivity = yield* readPublishAgentActivityEnabled.pipe(
842 Effect.orElseSucceed(() => false),
843 );
844 if (!publishAgentActivity) {
845 yield* Effect.logDebug("agent activity snapshot skipped; publication disabled");
846 return "disabled" as const;
847 }
848 const relayConfig = yield* readRelayConfig.pipe(Effect.orElseSucceed(() => null));
849 if (!relayConfig) {
850 yield* Effect.logDebug("agent activity snapshot skipped; relay link credentials unavailable");
851 return "unlinked" as const;
852 }
853 const environmentId = yield* serverEnvironment.getEnvironmentId;
854 const projectSnapshot = yield* projects.snapshot;
855 const activitySnapshot = yield* threads.getAgentAwarenessSnapshot();
856 // Live children aren't listed, but a waiting one makes its top-level thread wait.
857 const liveChildren = yield* Effect.forEach(
858 (yield* threads.listActiveShellIndex()).filter(
859 (thread) => thread.lineage.relationshipToParent === "subagent" && thread.mayBeLive,
860 ),
861 (thread) => readAwarenessThread(thread.id),
862 );
863 const activeThreadIds = resolveAgentAwarenessRelayActiveThreadIds({
864 environmentId,
865 startedAt,
866 projects: projectSnapshot.projects,
867 threads: [...activitySnapshot.threads, ...liveChildren.filter((thread) => thread !== null)],
868 });
869 if (activeThreadIds.length === 0) {
870 yield* Effect.logDebug("agent activity snapshot has no publishable threads");
871 return "published" as const;
872 }
873 yield* Effect.logInfo("publishing active agent activity snapshot", {
874 count: activeThreadIds.length,
875 });
876 yield* Effect.forEach(activeThreadIds, enqueueThreadPublish, { discard: true });
877 yield* worker.drain;
878 return "published" as const;
879 });
880
881 // Many environments never link, so while unlinked the retry backs off from 5
882 // s to 60 s. A linked environment keeps the 5 s retry, because `t3 connect
883 // publish` can turn publishing on from another process. Keep this untraced:
884 // an unlinked environment can retry for the lifetime of the process.
885 const publishActiveThreadsWhenConfigured = Effect.fnUntraced(function* (
886 logEnabledWhenReady: boolean,
887 ) {
888 let unlinkedRetryDelayMs = 5_000;
889 while (true) {
890 const result = yield* publishActiveThreadsUnsafe.pipe(
891 Effect.orElseSucceed(() => "failed" as const),
892 );
893 if (result === "published") {
894 if (logEnabledWhenReady) {
895 const relayConfig = yield* readRelayConfig.pipe(Effect.orElseSucceed(() => null));
896 yield* Effect.logInfo("agent activity publishing enabled after link reconciliation", {
897 relayUrl: relayConfig?.url,
898 });
899 }
900 logEnabledWhenReady = false;
901 unlinkedRetryDelayMs = 5_000;
902 yield* Queue.take(catchUpRequests);
903 continue;
904 }
905 const retryDelayMs = result === "unlinked" ? unlinkedRetryDelayMs : 5_000;
906 yield* Effect.race(Effect.sleep(retryDelayMs), Queue.take(catchUpRequests));
907 if (result === "unlinked") {
908 unlinkedRetryDelayMs = Math.min(unlinkedRetryDelayMs * 2, 60_000);
909 }
910 }
911 });
912
913 schedulePublishConfirm = (threadId) =>
914 Effect.forkIn(
915 Effect.sleep("5 seconds").pipe(
916 Effect.andThen(worker.enqueue(threadId)),
917 Effect.catchCause((cause) =>
918 Effect.logWarning("deferred agent activity confirmation failed", {
919 threadId,
920 cause: Cause.pretty(cause),
921 }),
922 ),
923 ),
924 scope,
925 ).pipe(Effect.asVoid);
926
927 const start: AgentAwarenessRelay["Service"]["start"] = Effect.fn("AgentAwarenessRelay.start")(
928 function* () {
929 const [relayConfig, publishEnabled] = yield* Effect.all([
930 readRelayConfig.pipe(Effect.orElseSucceed(() => null)),
931 readPublishAgentActivityEnabled.pipe(Effect.orElseSucceed(() => false)),
932 ]);
933 const startupState = resolveAgentActivityPublishingStartupState({
934 relayConfigured: relayConfig !== null,
935 publishEnabled,
936 });
937 switch (startupState) {
938 case "waiting-for-link":
939 yield* Effect.logInfo(
940 "agent activity publishing standby; waiting for T3 Connect link reconciliation",
941 );
942 break;
943 case "disabled":
944 yield* Effect.logInfo("agent activity publishing disabled by T3 Connect configuration");
945 break;
946 case "enabled":
947 yield* Effect.logInfo("agent activity publishing enabled", {
948 relayUrl: relayConfig?.url,
949 });
950 break;
951 }
952 yield* forkParked(
953 Effect.sleep("1 second").pipe(
954 Effect.andThen(publishActiveThreadsWhenConfigured(startupState !== "enabled")),
955 ),
956 );
957 yield* forkParked(
958 Stream.runForEach(threads.streamDomainEvents, (event) => {
959 const threadId = eventThreadId(event);
960 if (!shouldPublishAgentAwarenessEvent(event)) {
961 return Effect.void;
962 }
963 return Effect.logDebug("agent activity publishing queued thread publish", {
964 eventType: event.type,
965 threadId,
966 }).pipe(Effect.andThen(enqueueThreadPublish(threadId)));
967 }),
968 );
969 },
970 );
971
972 return AgentAwarenessRelay.of({
973 publishThread,
974 drain: worker.drain,
975 requestCatchUp: Effect.fn("AgentAwarenessRelay.requestCatchUp")(function () {
976 return Queue.offer(catchUpRequests, undefined).pipe(Effect.asVoid);
977 }),
978 start,
979 });
980});
981
982export const layer = Layer.effect(AgentAwarenessRelay, make);