import type { EnvironmentId, OrchestrationEvent, OrchestrationProjectShell, OrchestrationThreadShell, ThreadId,} from "@t3tools/contracts";import { RelayApi, type RelayAgentActivityPublishProofPayload, type RelayAgentActivityState,} from "@t3tools/contracts/relay";import { projectThreadAwareness } from "@t3tools/shared/agentAwareness";import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker";import { withRelayClientTracing } from "@t3tools/shared/relayTracing";import { normalizeRelayIssuer, RELAY_ACTIVITY_PUBLISH_TYP, signRelayJwt,} from "@t3tools/shared/relayJwt";import * as Cause from "effect/Cause";import * as Context from "effect/Context";import * as Crypto from "effect/Crypto";import * as DateTime from "effect/DateTime";import * as Effect from "effect/Effect";import * as Layer from "effect/Layer";import * as Option from "effect/Option";import * as Ref from "effect/Ref";import type * as Scope from "effect/Scope";import * as Stream from "effect/Stream";import * as FetchHttpClient from "effect/unstable/http/FetchHttpClient";import * as HttpClient from "effect/unstable/http/HttpClient";import * as HttpClientRequest from "effect/unstable/http/HttpClientRequest";import * as HttpApiClient from "effect/unstable/httpapi/HttpApiClient"; import * as ServerSecretStore from "../auth/ServerSecretStore.ts";import { isAgentActivityPublishingEnabledValue, PUBLISH_AGENT_ACTIVITY_SECRET, RELAY_ENVIRONMENT_CREDENTIAL_SECRET, RELAY_ISSUER_SECRET, RELAY_URL_SECRET,} from "../cloud/config.ts";import { getOrCreateEnvironmentKeyPairFromSecretStore } from "../cloud/environmentKeys.ts";import * as ServerEnvironment from "../environment/ServerEnvironment.ts";import * as OrchestrationEngine from "../orchestration/Services/OrchestrationEngine.ts";import * as ProjectionSnapshotQuery from "../orchestration/Services/ProjectionSnapshotQuery.ts";import { forkParked } from "../serverActivation.ts"; export class AgentAwarenessRelay extends Context.Service< AgentAwarenessRelay, { readonly publishThread: (threadId: ThreadId) => Effect.Effect<void>; readonly start: () => Effect.Effect<void, never, Scope.Scope>; }>()("t3/relay/AgentAwarenessRelay") {} export function eventThreadId(event: OrchestrationEvent): ThreadId | null { const payload = event.payload as { readonly threadId?: unknown }; if (typeof payload.threadId === "string") { return payload.threadId as ThreadId; } if (event.aggregateKind === "thread" && typeof event.aggregateId === "string") { return event.aggregateId as ThreadId; } return null;} export function shouldPublishAgentAwarenessEvent(event: OrchestrationEvent): boolean { if (event.metadata.historyImport === true) { return false; } switch (event.type) { case "thread.message-sent": case "thread.turn-start-requested": // These events express intent to start work, but the shell still contains // the previous turn's terminal state until the provider acknowledges the // new turn. Publishing that snapshot can queue a fresh "Done" alert just // before the real running state arrives. Provider lifecycle events publish // the authoritative starting/running state instead. return false; case "thread.proposed-plan-upserted": case "thread.runtime-mode-set": case "thread.interaction-mode-set": return false; case "thread.activity-appended": return ( event.payload.activity.kind === "approval.requested" || event.payload.activity.kind === "approval.resolved" || event.payload.activity.kind === "provider.approval.respond.failed" || event.payload.activity.kind === "user-input.requested" || event.payload.activity.kind === "user-input.resolved" || event.payload.activity.kind === "runtime.error" ); default: return true; }} export function agentAwarenessPublishIdentity(state: RelayAgentActivityState | null): string { if (state === null) { return "null"; } const { updatedAt: _updatedAt, ...meaningfulState } = state; return JSON.stringify(meaningfulState);} export function resolveAgentActivityPublishingStartupState(input: { readonly relayConfigured: boolean; readonly publishEnabled: boolean;}): "waiting-for-link" | "disabled" | "enabled" { if (!input.relayConfigured) { return "waiting-for-link"; } return input.publishEnabled ? "enabled" : "disabled";} const RELAY_AGENT_ACTIVITY_DETAIL_MAX_LENGTH = 160;const REDACTED_RELAY_AGENT_FAILURE_DETAIL = "The agent run failed."; export function sanitizeRelayAgentActivityState( state: RelayAgentActivityState | null,): RelayAgentActivityState | null { if (state === null) { return null; } const { detail: _detail, ...rest } = state; const detail = (state.phase === "failed" ? REDACTED_RELAY_AGENT_FAILURE_DETAIL : state.detail) ?.trim() .slice(0, RELAY_AGENT_ACTIVITY_DETAIL_MAX_LENGTH) .trim(); return detail ? { ...rest, detail } : rest;} function relayEnvironmentClient(token: string) { return HttpClient.mapRequest(HttpClientRequest.setHeader("authorization", `Bearer ${token}`));} function deliveryStats( deliveries: ReadonlyArray<{ readonly ok: boolean; readonly queued?: boolean | undefined; readonly kind: string; readonly apnsStatus?: number | null; readonly apnsReason?: string | null; }>,) { let queued = 0; let successful = 0; let failed = 0; const failedReasons: string[] = []; const kinds = new Set<string>(); for (const delivery of deliveries) { kinds.add(delivery.kind); if (delivery.queued) { queued += 1; continue; } if (delivery.ok) { successful += 1; continue; } failed += 1; failedReasons.push(`${delivery.apnsStatus ?? "transport"}:${delivery.apnsReason ?? "unknown"}`); } return { total: deliveries.length, queued, successful, failed, kinds: [...kinds], failedReasons, };} export function signRelayAgentActivityPublishProof(input: { readonly privateKey: string; readonly payload: RelayAgentActivityPublishProofPayload;}) { return signRelayJwt({ privateKey: input.privateKey, typ: RELAY_ACTIVITY_PUBLISH_TYP, payload: input.payload, });} const makePublishProof = Effect.fn("makePublishProof")(function* (input: { readonly privateKey: string; readonly relayIssuer: string; readonly environmentId: string; readonly threadId: ThreadId; readonly state: RelayAgentActivityState | null; readonly jti: string;}) { const now = yield* DateTime.now; const expiresAt = DateTime.add(now, { minutes: 5 }); const payload = { iss: `t3-env:${input.environmentId}`, aud: normalizeRelayIssuer(input.relayIssuer), sub: input.environmentId, jti: input.jti, iat: Math.floor(now.epochMilliseconds / 1_000), exp: Math.floor(expiresAt.epochMilliseconds / 1_000), environmentId: input.environmentId as RelayAgentActivityPublishProofPayload["environmentId"], threadId: input.threadId, state: input.state, } satisfies RelayAgentActivityPublishProofPayload; return yield* signRelayAgentActivityPublishProof({ privateKey: input.privateKey, payload });}); // Compact, log-safe view of the fields the awareness phase ladder reads.function describeThreadShellForAwareness( thread: Option.Option<OrchestrationThreadShell>,): Record<string, unknown> { if (Option.isNone(thread)) { return { found: false }; } const shell = thread.value; return { found: true, sessionStatus: shell.session?.status ?? null, sessionActiveTurnId: shell.session?.activeTurnId ?? null, latestTurnId: shell.latestTurn?.turnId ?? null, latestTurnState: shell.latestTurn?.state ?? null, latestTurnCompletedAt: shell.latestTurn?.completedAt ?? null, hasPendingApprovals: shell.hasPendingApprovals, hasPendingUserInput: shell.hasPendingUserInput, };} export function resolveAgentAwarenessRelayPublishSnapshot(input: { readonly environmentId: EnvironmentId; readonly threadId: ThreadId; readonly thread: Option.Option<OrchestrationThreadShell>; readonly project: Option.Option<OrchestrationProjectShell>;}): { readonly projectId: string | null; readonly state: RelayAgentActivityState | null; readonly reason: "snapshot" | "thread-not-found" | "project-not-found";} { if (Option.isNone(input.thread)) { return { projectId: null, state: null, reason: "thread-not-found", }; } if (Option.isNone(input.project)) { return { projectId: input.thread.value.projectId, state: null, reason: "project-not-found", }; } return { projectId: input.thread.value.projectId, state: sanitizeRelayAgentActivityState( projectThreadAwareness({ environmentId: input.environmentId, project: input.project.value, thread: input.thread.value, }), ), reason: "snapshot", };} export function resolveAgentAwarenessRelayActiveThreadIds(input: { readonly environmentId: EnvironmentId; readonly projects: ReadonlyArray<Pick<OrchestrationProjectShell, "id" | "title">>; readonly threads: ReadonlyArray<OrchestrationThreadShell>;}): ReadonlyArray<ThreadId> { const projectById = new Map(input.projects.map((project) => [project.id, project])); return input.threads .filter((thread) => { const project = projectById.get(thread.projectId); if (!project) { return false; } return ( projectThreadAwareness({ environmentId: input.environmentId, project, thread, }) !== null ); }) .map((thread) => thread.id);} /** @public Service construction is part of the canonical Effect module API. */export const make = Effect.gen(function* () { const secrets = yield* ServerSecretStore.ServerSecretStore; const serverEnvironment = yield* ServerEnvironment.ServerEnvironment; const snapshotQuery = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery; const orchestrationEngine = yield* OrchestrationEngine.OrchestrationEngineService; const crypto = yield* Crypto.Crypto; const cloudLinkKeyPair = yield* getOrCreateEnvironmentKeyPairFromSecretStore(secrets); const activeSnapshotPublishedRef = yield* Ref.make(false); const publishedStateByThreadRef = yield* Ref.make(new Map<ThreadId, string>()); const readSecretString = (name: string) => secrets .get(name) .pipe( Effect.map((bytes) => Option.isSome(bytes) ? new TextDecoder().decode(bytes.value) : null, ), ); const readRelayConfig = Effect.gen(function* () { const [url, issuer, environmentCredential] = yield* Effect.all([ readSecretString(RELAY_URL_SECRET), readSecretString(RELAY_ISSUER_SECRET), readSecretString(RELAY_ENVIRONMENT_CREDENTIAL_SECRET), ]); return url && environmentCredential ? { url, issuer: issuer ?? url, environmentCredential } : null; }); const readPublishAgentActivityEnabled = readSecretString(PUBLISH_AGENT_ACTIVITY_SECRET).pipe( Effect.map(isAgentActivityPublishingEnabledValue), ); const makeRelayClient = (relayConfig: { readonly url: string; readonly environmentCredential: string; }) => HttpApiClient.make(RelayApi, { baseUrl: relayConfig.url, transformClient: relayEnvironmentClient(relayConfig.environmentCredential), }).pipe(Effect.provide(FetchHttpClient.layer)); // Deadlines for publishes that need confirmation (tombstones and // first-state completions). The confirming publish is re-enqueued through // the same drainable worker as every other publish, so a confirmed // tombstone can never race an in-flight live update; a recovered state // clears the deadline. Assigned after the worker exists. const publishConfirmDeadlines = new Map<ThreadId, number>(); let schedulePublishConfirm: (threadId: ThreadId) => Effect.Effect<void> = () => Effect.void; const publishThreadUnsafe = Effect.fn("publishThreadUnsafe")(function* (threadId: ThreadId) { const publishAgentActivity = yield* readPublishAgentActivityEnabled.pipe( Effect.orElseSucceed(() => false), ); if (!publishAgentActivity) { yield* Effect.logDebug("agent activity publish skipped; publication disabled", { threadId, }); return; } const relayConfig = yield* readRelayConfig.pipe(Effect.orElseSucceed(() => null)); if (!relayConfig) { yield* Effect.logDebug("agent activity publish skipped; relay link credentials unavailable", { threadId, }); return; } const relayClient = yield* makeRelayClient(relayConfig); const environmentId = yield* serverEnvironment.getEnvironmentId; const publishState = (input: { readonly projectId: string | null; readonly state: RelayAgentActivityState | null; readonly reason: string; }) => Effect.gen(function* () { const proof = yield* makePublishProof({ privateKey: cloudLinkKeyPair.privateKey, relayIssuer: relayConfig.issuer, environmentId, threadId, state: input.state, jti: yield* crypto.randomUUIDv4, }); yield* Effect.logInfo("publishing agent activity for thread", { environmentId, threadId, projectId: input.projectId, statePhase: input.state?.phase ?? null, hasState: input.state !== null, reason: input.reason, }); const response = yield* relayClient.server.publishAgentActivity({ params: { environmentId, threadId, }, payload: { state: input.state, proof, }, }); yield* Effect.logInfo("agent activity publish completed", { environmentId, threadId, ok: response.ok, deliveries: deliveryStats(response.deliveries), }); }); const thread = yield* snapshotQuery.getThreadShellById(threadId); const project = Option.isSome(thread) ? yield* snapshotQuery.getProjectShellById(thread.value.projectId) : Option.none<OrchestrationProjectShell>(); const snapshot = resolveAgentAwarenessRelayPublishSnapshot({ environmentId, threadId, thread, project, }); const publishIdentity = agentAwarenessPublishIdentity(snapshot.state); const publishedStateByThread = yield* Ref.get(publishedStateByThreadRef); if (publishedStateByThread.get(threadId) === publishIdentity) { // The projection is back at (or never left) the last published state, so // any pending deferred confirmation is moot. Leaving the deadline in // place would let a much later transient null find it already expired // and publish a tombstone immediately, skipping the deferral window. publishConfirmDeadlines.delete(threadId); yield* Effect.logDebug("agent activity publish skipped; projected state unchanged", { environmentId, threadId, reason: snapshot.reason, }); return; } // Two projections need confirmation before publishing, because both can // appear transiently while the projector is mid-write and publishing them // immediately is destructive or noisy: // - null (tombstone) while the previous published state was live: deletes // the thread from every armed card mid-conversation. // - completed as the thread's FIRST published state: sessions boot at // "ready" before their first turn, which projects as completed for an // instant and sends a spurious Done notification at thread birth. // Defer, schedule a re-publish through the ordinary worker queue, and // only publish if the projection still holds when it drains. const requiresConfirmation = (snapshot.state === null && publishedStateByThread.get(threadId) !== agentAwarenessPublishIdentity(null)) || (snapshot.state?.phase === "completed" && !publishedStateByThread.has(threadId)); if (requiresConfirmation) { const nowMs = (yield* DateTime.now).epochMilliseconds; const deadline = publishConfirmDeadlines.get(threadId); if (deadline === undefined) { publishConfirmDeadlines.set(threadId, nowMs + 5_000); yield* Effect.logInfo("agent activity publish deferred pending confirmation", { environmentId, threadId, reason: snapshot.reason, statePhase: snapshot.state?.phase ?? null, shell: describeThreadShellForAwareness(thread), }); yield* schedulePublishConfirm(threadId); return; } if (nowMs < deadline) { return; } publishConfirmDeadlines.delete(threadId); yield* Effect.logInfo("agent activity deferred publish confirmed", { environmentId, threadId, reason: snapshot.reason, statePhase: snapshot.state?.phase ?? null, shell: describeThreadShellForAwareness(thread), }); } else { publishConfirmDeadlines.delete(threadId); } if (snapshot.reason === "thread-not-found") { yield* Effect.logDebug("publishing agent activity tombstone; thread not found", { environmentId, threadId, }); } else if (snapshot.reason === "project-not-found") { yield* Effect.logDebug("publishing agent activity tombstone; project not found", { environmentId, threadId, projectId: snapshot.projectId, }); } yield* publishState({ projectId: snapshot.projectId, state: snapshot.state, reason: snapshot.reason, }); yield* Ref.update(publishedStateByThreadRef, (publishedStates) => { const nextPublishedStates = new Map(publishedStates); nextPublishedStates.set(threadId, publishIdentity); return nextPublishedStates; });Lines 1–500 of 641. Load more
import type { EnvironmentId, OrchestrationEvent, OrchestrationProjectShell, OrchestrationThreadShell, ThreadId, } from "@t3tools/contracts"; import { RelayApi, type RelayAgentActivityPublishProofPayload, type RelayAgentActivityState, } from "@t3tools/contracts/relay"; import { projectThreadAwareness } from "@t3tools/shared/agentAwareness"; import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker"; import { withRelayClientTracing } from "@t3tools/shared/relayTracing"; import { normalizeRelayIssuer, RELAY_ACTIVITY_PUBLISH_TYP, signRelayJwt, } from "@t3tools/shared/relayJwt"; import * as Cause from "effect/Cause"; import * as Context from "effect/Context"; import * as Crypto from "effect/Crypto"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Ref from "effect/Ref"; import type * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; import * as FetchHttpClient from "effect/unstable/http/FetchHttpClient"; import * as HttpClient from "effect/unstable/http/HttpClient"; import * as HttpClientRequest from "effect/unstable/http/HttpClientRequest"; import * as HttpApiClient from "effect/unstable/httpapi/HttpApiClient"; import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; import { isAgentActivityPublishingEnabledValue, PUBLISH_AGENT_ACTIVITY_SECRET, RELAY_ENVIRONMENT_CREDENTIAL_SECRET, RELAY_ISSUER_SECRET, RELAY_URL_SECRET, } from "../cloud/config.ts"; import { getOrCreateEnvironmentKeyPairFromSecretStore } from "../cloud/environmentKeys.ts"; import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; import * as OrchestrationEngine from "../orchestration/Services/OrchestrationEngine.ts"; import * as ProjectionSnapshotQuery from "../orchestration/Services/ProjectionSnapshotQuery.ts"; import { forkParked } from "../serverActivation.ts"; export class AgentAwarenessRelay extends Context.Service< AgentAwarenessRelay, { readonly publishThread: (threadId: ThreadId) => Effect.Effect<void>; readonly start: () => Effect.Effect<void, never, Scope.Scope>; } >()("t3/relay/AgentAwarenessRelay") {} export function eventThreadId(event: OrchestrationEvent): ThreadId | null { const payload = event.payload as { readonly threadId?: unknown }; if (typeof payload.threadId === "string") { return payload.threadId as ThreadId; } if (event.aggregateKind === "thread" && typeof event.aggregateId === "string") { return event.aggregateId as ThreadId; } return null; } export function shouldPublishAgentAwarenessEvent(event: OrchestrationEvent): boolean { if (event.metadata.historyImport === true) { return false; } switch (event.type) { case "thread.message-sent": case "thread.turn-start-requested": // These events express intent to start work, but the shell still contains // the previous turn's terminal state until the provider acknowledges the // new turn. Publishing that snapshot can queue a fresh "Done" alert just // before the real running state arrives. Provider lifecycle events publish // the authoritative starting/running state instead. return false; case "thread.proposed-plan-upserted": case "thread.runtime-mode-set": case "thread.interaction-mode-set": return false; case "thread.activity-appended": return ( event.payload.activity.kind === "approval.requested" || event.payload.activity.kind === "approval.resolved" || event.payload.activity.kind === "provider.approval.respond.failed" || event.payload.activity.kind === "user-input.requested" || event.payload.activity.kind === "user-input.resolved" || event.payload.activity.kind === "runtime.error" ); default: return true; } } export function agentAwarenessPublishIdentity(state: RelayAgentActivityState | null): string { if (state === null) { return "null"; } const { updatedAt: _updatedAt, ...meaningfulState } = state; return JSON.stringify(meaningfulState); } export function resolveAgentActivityPublishingStartupState(input: { readonly relayConfigured: boolean; readonly publishEnabled: boolean; }): "waiting-for-link" | "disabled" | "enabled" { if (!input.relayConfigured) { return "waiting-for-link"; } return input.publishEnabled ? "enabled" : "disabled"; } const RELAY_AGENT_ACTIVITY_DETAIL_MAX_LENGTH = 160; const REDACTED_RELAY_AGENT_FAILURE_DETAIL = "The agent run failed."; export function sanitizeRelayAgentActivityState( state: RelayAgentActivityState | null, ): RelayAgentActivityState | null { if (state === null) { return null; } const { detail: _detail, ...rest } = state; const detail = (state.phase === "failed" ? REDACTED_RELAY_AGENT_FAILURE_DETAIL : state.detail) ?.trim() .slice(0, RELAY_AGENT_ACTIVITY_DETAIL_MAX_LENGTH) .trim(); return detail ? { ...rest, detail } : rest; } function relayEnvironmentClient(token: string) { return HttpClient.mapRequest(HttpClientRequest.setHeader("authorization", `Bearer ${token}`)); } function deliveryStats( deliveries: ReadonlyArray<{ readonly ok: boolean; readonly queued?: boolean | undefined; readonly kind: string; readonly apnsStatus?: number | null; readonly apnsReason?: string | null; }>, ) { let queued = 0; let successful = 0; let failed = 0; const failedReasons: string[] = []; const kinds = new Set<string>(); for (const delivery of deliveries) { kinds.add(delivery.kind); if (delivery.queued) { queued += 1; continue; } if (delivery.ok) { successful += 1; continue; } failed += 1; failedReasons.push(`${delivery.apnsStatus ?? "transport"}:${delivery.apnsReason ?? "unknown"}`); } return { total: deliveries.length, queued, successful, failed, kinds: [...kinds], failedReasons, }; } export function signRelayAgentActivityPublishProof(input: { readonly privateKey: string; readonly payload: RelayAgentActivityPublishProofPayload; }) { return signRelayJwt({ privateKey: input.privateKey, typ: RELAY_ACTIVITY_PUBLISH_TYP, payload: input.payload, }); } const makePublishProof = Effect.fn("makePublishProof")(function* (input: { readonly privateKey: string; readonly relayIssuer: string; readonly environmentId: string; readonly threadId: ThreadId; readonly state: RelayAgentActivityState | null; readonly jti: string; }) { const now = yield* DateTime.now; const expiresAt = DateTime.add(now, { minutes: 5 }); const payload = { iss: `t3-env:${input.environmentId}`, aud: normalizeRelayIssuer(input.relayIssuer), sub: input.environmentId, jti: input.jti, iat: Math.floor(now.epochMilliseconds / 1_000), exp: Math.floor(expiresAt.epochMilliseconds / 1_000), environmentId: input.environmentId as RelayAgentActivityPublishProofPayload["environmentId"], threadId: input.threadId, state: input.state, } satisfies RelayAgentActivityPublishProofPayload; return yield* signRelayAgentActivityPublishProof({ privateKey: input.privateKey, payload }); }); // Compact, log-safe view of the fields the awareness phase ladder reads. function describeThreadShellForAwareness( thread: Option.Option<OrchestrationThreadShell>, ): Record<string, unknown> { if (Option.isNone(thread)) { return { found: false }; } const shell = thread.value; return { found: true, sessionStatus: shell.session?.status ?? null, sessionActiveTurnId: shell.session?.activeTurnId ?? null, latestTurnId: shell.latestTurn?.turnId ?? null, latestTurnState: shell.latestTurn?.state ?? null, latestTurnCompletedAt: shell.latestTurn?.completedAt ?? null, hasPendingApprovals: shell.hasPendingApprovals, hasPendingUserInput: shell.hasPendingUserInput, }; } export function resolveAgentAwarenessRelayPublishSnapshot(input: { readonly environmentId: EnvironmentId; readonly threadId: ThreadId; readonly thread: Option.Option<OrchestrationThreadShell>; readonly project: Option.Option<OrchestrationProjectShell>; }): { readonly projectId: string | null; readonly state: RelayAgentActivityState | null; readonly reason: "snapshot" | "thread-not-found" | "project-not-found"; } { if (Option.isNone(input.thread)) { return { projectId: null, state: null, reason: "thread-not-found", }; } if (Option.isNone(input.project)) { return { projectId: input.thread.value.projectId, state: null, reason: "project-not-found", }; } return { projectId: input.thread.value.projectId, state: sanitizeRelayAgentActivityState( projectThreadAwareness({ environmentId: input.environmentId, project: input.project.value, thread: input.thread.value, }), ), reason: "snapshot", }; } export function resolveAgentAwarenessRelayActiveThreadIds(input: { readonly environmentId: EnvironmentId; readonly projects: ReadonlyArray<Pick<OrchestrationProjectShell, "id" | "title">>; readonly threads: ReadonlyArray<OrchestrationThreadShell>; }): ReadonlyArray<ThreadId> { const projectById = new Map(input.projects.map((project) => [project.id, project])); return input.threads .filter((thread) => { const project = projectById.get(thread.projectId); if (!project) { return false; } return ( projectThreadAwareness({ environmentId: input.environmentId, project, thread, }) !== null ); }) .map((thread) => thread.id); } /** @public Service construction is part of the canonical Effect module API. */ export const make = Effect.gen(function* () { const secrets = yield* ServerSecretStore.ServerSecretStore; const serverEnvironment = yield* ServerEnvironment.ServerEnvironment; const snapshotQuery = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery; const orchestrationEngine = yield* OrchestrationEngine.OrchestrationEngineService; const crypto = yield* Crypto.Crypto; const cloudLinkKeyPair = yield* getOrCreateEnvironmentKeyPairFromSecretStore(secrets); const activeSnapshotPublishedRef = yield* Ref.make(false); const publishedStateByThreadRef = yield* Ref.make(new Map<ThreadId, string>()); const readSecretString = (name: string) => secrets .get(name) .pipe( Effect.map((bytes) => Option.isSome(bytes) ? new TextDecoder().decode(bytes.value) : null, ), ); const readRelayConfig = Effect.gen(function* () { const [url, issuer, environmentCredential] = yield* Effect.all([ readSecretString(RELAY_URL_SECRET), readSecretString(RELAY_ISSUER_SECRET), readSecretString(RELAY_ENVIRONMENT_CREDENTIAL_SECRET), ]); return url && environmentCredential ? { url, issuer: issuer ?? url, environmentCredential } : null; }); const readPublishAgentActivityEnabled = readSecretString(PUBLISH_AGENT_ACTIVITY_SECRET).pipe( Effect.map(isAgentActivityPublishingEnabledValue), ); const makeRelayClient = (relayConfig: { readonly url: string; readonly environmentCredential: string; }) => HttpApiClient.make(RelayApi, { baseUrl: relayConfig.url, transformClient: relayEnvironmentClient(relayConfig.environmentCredential), }).pipe(Effect.provide(FetchHttpClient.layer)); // Deadlines for publishes that need confirmation (tombstones and // first-state completions). The confirming publish is re-enqueued through // the same drainable worker as every other publish, so a confirmed // tombstone can never race an in-flight live update; a recovered state // clears the deadline. Assigned after the worker exists. const publishConfirmDeadlines = new Map<ThreadId, number>(); let schedulePublishConfirm: (threadId: ThreadId) => Effect.Effect<void> = () => Effect.void; const publishThreadUnsafe = Effect.fn("publishThreadUnsafe")(function* (threadId: ThreadId) { const publishAgentActivity = yield* readPublishAgentActivityEnabled.pipe( Effect.orElseSucceed(() => false), ); if (!publishAgentActivity) { yield* Effect.logDebug("agent activity publish skipped; publication disabled", { threadId, }); return; } const relayConfig = yield* readRelayConfig.pipe(Effect.orElseSucceed(() => null)); if (!relayConfig) { yield* Effect.logDebug("agent activity publish skipped; relay link credentials unavailable", { threadId, }); return; } const relayClient = yield* makeRelayClient(relayConfig); const environmentId = yield* serverEnvironment.getEnvironmentId; const publishState = (input: { readonly projectId: string | null; readonly state: RelayAgentActivityState | null; readonly reason: string; }) => Effect.gen(function* () { const proof = yield* makePublishProof({ privateKey: cloudLinkKeyPair.privateKey, relayIssuer: relayConfig.issuer, environmentId, threadId, state: input.state, jti: yield* crypto.randomUUIDv4, }); yield* Effect.logInfo("publishing agent activity for thread", { environmentId, threadId, projectId: input.projectId, statePhase: input.state?.phase ?? null, hasState: input.state !== null, reason: input.reason, }); const response = yield* relayClient.server.publishAgentActivity({ params: { environmentId, threadId, }, payload: { state: input.state, proof, }, }); yield* Effect.logInfo("agent activity publish completed", { environmentId, threadId, ok: response.ok, deliveries: deliveryStats(response.deliveries), }); }); const thread = yield* snapshotQuery.getThreadShellById(threadId); const project = Option.isSome(thread) ? yield* snapshotQuery.getProjectShellById(thread.value.projectId) : Option.none<OrchestrationProjectShell>(); const snapshot = resolveAgentAwarenessRelayPublishSnapshot({ environmentId, threadId, thread, project, }); const publishIdentity = agentAwarenessPublishIdentity(snapshot.state); const publishedStateByThread = yield* Ref.get(publishedStateByThreadRef); if (publishedStateByThread.get(threadId) === publishIdentity) { // The projection is back at (or never left) the last published state, so // any pending deferred confirmation is moot. Leaving the deadline in // place would let a much later transient null find it already expired // and publish a tombstone immediately, skipping the deferral window. publishConfirmDeadlines.delete(threadId); yield* Effect.logDebug("agent activity publish skipped; projected state unchanged", { environmentId, threadId, reason: snapshot.reason, }); return; } // Two projections need confirmation before publishing, because both can // appear transiently while the projector is mid-write and publishing them // immediately is destructive or noisy: // - null (tombstone) while the previous published state was live: deletes // the thread from every armed card mid-conversation. // - completed as the thread's FIRST published state: sessions boot at // "ready" before their first turn, which projects as completed for an // instant and sends a spurious Done notification at thread birth. // Defer, schedule a re-publish through the ordinary worker queue, and // only publish if the projection still holds when it drains. const requiresConfirmation = (snapshot.state === null && publishedStateByThread.get(threadId) !== agentAwarenessPublishIdentity(null)) || (snapshot.state?.phase === "completed" && !publishedStateByThread.has(threadId)); if (requiresConfirmation) { const nowMs = (yield* DateTime.now).epochMilliseconds; const deadline = publishConfirmDeadlines.get(threadId); if (deadline === undefined) { publishConfirmDeadlines.set(threadId, nowMs + 5_000); yield* Effect.logInfo("agent activity publish deferred pending confirmation", { environmentId, threadId, reason: snapshot.reason, statePhase: snapshot.state?.phase ?? null, shell: describeThreadShellForAwareness(thread), }); yield* schedulePublishConfirm(threadId); return; } if (nowMs < deadline) { return; } publishConfirmDeadlines.delete(threadId); yield* Effect.logInfo("agent activity deferred publish confirmed", { environmentId, threadId, reason: snapshot.reason, statePhase: snapshot.state?.phase ?? null, shell: describeThreadShellForAwareness(thread), }); } else { publishConfirmDeadlines.delete(threadId); } if (snapshot.reason === "thread-not-found") { yield* Effect.logDebug("publishing agent activity tombstone; thread not found", { environmentId, threadId, }); } else if (snapshot.reason === "project-not-found") { yield* Effect.logDebug("publishing agent activity tombstone; project not found", { environmentId, threadId, projectId: snapshot.projectId, }); } yield* publishState({ projectId: snapshot.projectId, state: snapshot.state, reason: snapshot.reason, }); yield* Ref.update(publishedStateByThreadRef, (publishedStates) => { const nextPublishedStates = new Map(publishedStates); nextPublishedStates.set(threadId, publishIdentity); return nextPublishedStates; });