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