authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-09-23 19:17:38-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-23 19:17:38-07:00
loge4eb9977f0b02b29cb2ee361b0c433c685e87615
tree9644e705a7a749d988afbb9f31d67491bc20a738
parent894d33419dceb366eecda8e5e27c6fc043c3959e
signature Signed by PGP key B5690EEEBB952194

fix(server): stop replaying old agent alerts on restart (#13340)

2 files changed, 207 insertions(+), 7 deletions(-)

apps/server/src/relay/AgentAwarenessRelay.test.ts+178-1
...@@ -20,6 +20,7 @@ import { CommandId, ProviderInstanceId } from "@t3tools/contracts";...@@ -20,6 +20,7 @@ import { CommandId, ProviderInstanceId } from "@t3tools/contracts";
20import { RelayClientTracer } from "@t3tools/shared/relayTracing";20import { RelayClientTracer } from "@t3tools/shared/relayTracing";
21import { RELAY_ACTIVITY_PUBLISH_TYP, verifyRelayJwt } from "@t3tools/shared/relayJwt";21import { RELAY_ACTIVITY_PUBLISH_TYP, verifyRelayJwt } from "@t3tools/shared/relayJwt";
22import { describe, expect, it } from "@effect/vitest";22import { describe, expect, it } from "@effect/vitest";
23import * as DateTime from "effect/DateTime";
23import * as Deferred from "effect/Deferred";24import * as Deferred from "effect/Deferred";
24import * as Effect from "effect/Effect";25import * as Effect from "effect/Effect";
25import * as Layer from "effect/Layer";26import * as Layer from "effect/Layer";
...@@ -325,6 +326,9 @@ describe.sequential("signRelayAgentActivityPublishProof", () => {...@@ -325,6 +326,9 @@ describe.sequential("signRelayAgentActivityPublishProof", () => {
325 const projectId = "project-1" as ProjectId;326 const projectId = "project-1" as ProjectId;
326 const activeThreadId = "thread-active" as ThreadId;327 const activeThreadId = "thread-active" as ThreadId;
327 const idleThreadId = "thread-idle" as ThreadId;328 const idleThreadId = "thread-idle" as ThreadId;
329 const oldCompletedId = "thread-old-completed" as ThreadId;
330 const newCompletedId = "thread-new-completed" as ThreadId;
331 const freshMessageId = "thread-fresh-message" as ThreadId;
328332
329 const baseThread = {333 const baseThread = {
330 projectId,334 projectId,
...@@ -351,6 +355,7 @@ describe.sequential("signRelayAgentActivityPublishProof", () => {...@@ -351,6 +355,7 @@ describe.sequential("signRelayAgentActivityPublishProof", () => {
351 expect(355 expect(
352 AgentAwarenessRelay.resolveAgentAwarenessRelayActiveThreadIds({356 AgentAwarenessRelay.resolveAgentAwarenessRelayActiveThreadIds({
353 environmentId,357 environmentId,
358 startedAt: Date.parse(now),
354 projects: [359 projects: [
355 {360 {
356 id: projectId,361 id: projectId,
...@@ -374,6 +379,44 @@ describe.sequential("signRelayAgentActivityPublishProof", () => {...@@ -374,6 +379,44 @@ describe.sequential("signRelayAgentActivityPublishProof", () => {
374 ...baseThread,379 ...baseThread,
375 id: idleThreadId,380 id: idleThreadId,
376 },381 },
382 {
383 ...baseThread,
384 id: oldCompletedId,
385 latestTurn: {
386 turnId: "turn-old" as TurnId,
387 state: "completed",
388 requestedAt: "2026-05-24T00:00:00.000Z",
389 startedAt: "2026-05-24T00:00:00.000Z",
390 completedAt: "2026-05-24T00:01:00.000Z",
391 assistantMessageId: null,
392 },
393 },
394 {
395 ...baseThread,
396 id: newCompletedId,
397 latestTurn: {
398 turnId: "turn-new" as TurnId,
399 state: "completed",
400 requestedAt: "2026-05-25T00:00:01.000Z",
401 startedAt: "2026-05-25T00:00:01.000Z",
402 completedAt: "2026-05-25T00:00:02.000Z",
403 assistantMessageId: null,
404 },
405 },
406 {
407 ...baseThread,
408 id: freshMessageId,
409 latestUserMessageAt: "2026-05-25T00:00:01.000Z",
410 session: {
411 threadId: freshMessageId,
412 status: "ready",
413 providerName: "Codex",
414 runtimeMode: "full-access",
415 activeTurnId: null,
416 lastError: null,
417 updatedAt: "2026-05-25T00:00:02.000Z",
418 },
419 },
377 {420 {
378 ...baseThread,421 ...baseThread,
379 id: "thread-missing-project" as ThreadId,422 id: "thread-missing-project" as ThreadId,
...@@ -389,7 +432,7 @@ describe.sequential("signRelayAgentActivityPublishProof", () => {...@@ -389,7 +432,7 @@ describe.sequential("signRelayAgentActivityPublishProof", () => {
389 },432 },
390 ],433 ],
391 }),434 }),
392 ).toEqual([activeThreadId]);435 ).toEqual([activeThreadId, newCompletedId]);
393 });436 });
394437
395 it.effect("signs the activity publish JWT and rejects tampering", () =>438 it.effect("signs the activity publish JWT and rejects tampering", () =>
...@@ -797,4 +840,138 @@ describe.sequential("signRelayAgentActivityPublishProof", () => {...@@ -797,4 +840,138 @@ describe.sequential("signRelayAgentActivityPublishProof", () => {
797 }),840 }),
798 ),841 ),
799 );842 );
843
844 it.effect("does not alert for historical completions after startup", () =>
845 Effect.scoped(
846 Effect.gen(function* () {
847 const secrets = makeMemorySecretStore();
848 const now = yield* DateTime.now;
849 const old = DateTime.formatIso(DateTime.add(now, { days: -7 }));
850 const threadId = "thread-old" as ThreadId;
851 const projectId = "project-1" as ProjectId;
852 const environmentId = "env-1" as EnvironmentId;
853 const project = {
854 id: projectId,
855 title: "T3 Code",
856 workspaceRoot: "/workspace",
857 repositoryIdentity: null,
858 defaultModelSelection: null,
859 scripts: [],
860 createdAt: old,
861 updatedAt: old,
862 } satisfies OrchestrationProjectShell;
863 const completedTurn = {
864 turnId: "turn-1" as TurnId,
865 state: "completed",
866 requestedAt: old,
867 startedAt: old,
868 completedAt: old,
869 assistantMessageId: null,
870 } as const;
871 const completedThread = {
872 id: threadId,
873 projectId,
874 title: "Old task",
875 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
876 runtimeMode: "full-access",
877 interactionMode: "default",
878 branch: null,
879 worktreePath: null,
880 pullRequests: [],
881 latestTurn: completedTurn,
882 createdAt: old,
883 updatedAt: old,
884 archivedAt: null,
885 settledOverride: null,
886 settledAt: null,
887 session: null,
888 latestUserMessageAt: old,
889 hasPendingApprovals: false,
890 hasPendingUserInput: false,
891 hasActionableProposedPlan: false,
892 } satisfies OrchestrationThreadShell;
893 let currentThread: OrchestrationThreadShell | null = completedThread;
894 let publishes = 0;
895 const originalFetch = globalThis.fetch;
896 globalThis.fetch = (() => {
897 publishes += 1;
898 return Promise.resolve(Response.json({ ok: true, deliveries: [] }));
899 }) as unknown as typeof fetch;
900 yield* Effect.addFinalizer(() =>
901 Effect.sync(() => {
902 globalThis.fetch = originalFetch;
903 }),
904 );
905 yield* secrets.setString(RELAY_URL_SECRET, "https://relay.example.test");
906 yield* secrets.setString(RELAY_ENVIRONMENT_CREDENTIAL_SECRET, "relay-credential");
907 yield* secrets.setString(PUBLISH_AGENT_ACTIVITY_SECRET, "true");
908
909 const layer = Layer.mergeAll(
910 Layer.succeed(ServerSecretStore.ServerSecretStore, secrets.store),
911 Layer.succeed(ServerEnvironment.ServerEnvironment, {
912 getEnvironmentId: Effect.succeed(environmentId),
913 getDescriptor: Effect.die("unused descriptor"),
914 }),
915 Layer.succeed(OrchestrationEngineService, {} as OrchestrationEngineShape),
916 Layer.succeed(ProjectionSnapshotQuery, {
917 getThreadShellById: () => Effect.sync(() => Option.fromNullishOr(currentThread)),
918 getProjectShellById: () => Effect.succeed(Option.some(project)),
919 } as unknown as ProjectionSnapshotQueryShape),
920 );
921
922 yield* Effect.gen(function* () {
923 const relay = yield* AgentAwarenessRelay.AgentAwarenessRelay;
924 yield* relay.publishThread(threadId);
925 expect(publishes).toBe(0);
926
927 currentThread = {
928 ...completedThread,
929 latestTurn: null,
930 latestUserMessageAt: DateTime.formatIso(DateTime.add(now, { seconds: 1 })),
931 session: {
932 threadId,
933 status: "ready",
934 providerName: "Codex",
935 runtimeMode: "full-access",
936 activeTurnId: null,
937 lastError: null,
938 updatedAt: DateTime.formatIso(DateTime.add(now, { seconds: 2 })),
939 },
940 };
941 expect(
942 AgentAwarenessRelay.resolveAgentAwarenessRelayPublishSnapshot({
943 environmentId,
944 threadId,
945 thread: Option.some(currentThread),
946 project: Option.some(project),
947 }).state?.phase,
948 ).toBe("completed");
949 yield* relay.publishThread(threadId);
950 expect(publishes).toBe(0);
951
952 currentThread = {
953 ...completedThread,
954 session: {
955 threadId,
956 status: "error",
957 providerName: "Codex",
958 runtimeMode: "full-access",
959 activeTurnId: null,
960 lastError: "old failure",
961 updatedAt: old,
962 },
963 };
964 yield* relay.publishThread(threadId);
965 expect(publishes).toBe(0);
966 }).pipe(
967 Effect.provide(
968 AgentAwarenessRelay.layer.pipe(
969 Layer.provide(layer),
970 Layer.provideMerge(NodeServices.layer),
971 ),
972 ),
973 );
974 }),
975 ),
976 );
800});977});
apps/server/src/relay/AgentAwarenessRelay.ts+29-6
...@@ -267,8 +267,13 @@ export function resolveAgentAwarenessRelayPublishSnapshot(input: {...@@ -267,8 +267,13 @@ export function resolveAgentAwarenessRelayPublishSnapshot(input: {
267 };267 };
268}268}
269269
270function terminalWorkSinceStart(thread: OrchestrationThreadShell, startedAt: number): boolean {
271 return Date.parse(thread.latestTurn?.completedAt ?? "") > startedAt;
272}
273
270export function resolveAgentAwarenessRelayActiveThreadIds(input: {274export function resolveAgentAwarenessRelayActiveThreadIds(input: {
271 readonly environmentId: EnvironmentId;275 readonly environmentId: EnvironmentId;
276 readonly startedAt: number;
272 readonly projects: ReadonlyArray<Pick<OrchestrationProjectShell, "id" | "title">>;277 readonly projects: ReadonlyArray<Pick<OrchestrationProjectShell, "id" | "title">>;
273 readonly threads: ReadonlyArray<OrchestrationThreadShell>;278 readonly threads: ReadonlyArray<OrchestrationThreadShell>;
274}): ReadonlyArray<ThreadId> {279}): ReadonlyArray<ThreadId> {
...@@ -279,12 +284,16 @@ export function resolveAgentAwarenessRelayActiveThreadIds(input: {...@@ -279,12 +284,16 @@ export function resolveAgentAwarenessRelayActiveThreadIds(input: {
279 if (!project) {284 if (!project) {
280 return false;285 return false;
281 }286 }
287 const state = projectThreadAwareness({
288 environmentId: input.environmentId,
289 project,
290 thread,
291 });
282 return (292 return (
283 projectThreadAwareness({293 state !== null &&
284 environmentId: input.environmentId,294 (state.phase !== "completed" && state.phase !== "failed"
285 project,295 ? true
286 thread,296 : terminalWorkSinceStart(thread, input.startedAt))
287 }) !== null
288 );297 );
289 })298 })
290 .map((thread) => thread.id);299 .map((thread) => thread.id);
...@@ -298,6 +307,7 @@ export const make = Effect.gen(function* () {...@@ -298,6 +307,7 @@ export const make = Effect.gen(function* () {
298 const orchestrationEngine = yield* OrchestrationEngine.OrchestrationEngineService;307 const orchestrationEngine = yield* OrchestrationEngine.OrchestrationEngineService;
299 const crypto = yield* Crypto.Crypto;308 const crypto = yield* Crypto.Crypto;
300 const cloudLinkKeyPair = yield* getOrCreateEnvironmentKeyPairFromSecretStore(secrets);309 const cloudLinkKeyPair = yield* getOrCreateEnvironmentKeyPairFromSecretStore(secrets);
310 const startedAt = (yield* DateTime.now).epochMilliseconds;
301 const activeSnapshotPublishedRef = yield* Ref.make(false);311 const activeSnapshotPublishedRef = yield* Ref.make(false);
302 const publishedStateByThreadRef = yield* Ref.make(new Map<ThreadId, string>());312 const publishedStateByThreadRef = yield* Ref.make(new Map<ThreadId, string>());
303313
...@@ -417,6 +427,14 @@ export const make = Effect.gen(function* () {...@@ -417,6 +427,14 @@ export const make = Effect.gen(function* () {
417 });427 });
418 const publishIdentity = agentAwarenessPublishIdentity(snapshot.state);428 const publishIdentity = agentAwarenessPublishIdentity(snapshot.state);
419 const publishedStateByThread = yield* Ref.get(publishedStateByThreadRef);429 const publishedStateByThread = yield* Ref.get(publishedStateByThreadRef);
430 if (
431 (snapshot.state?.phase === "completed" || snapshot.state?.phase === "failed") &&
432 !publishedStateByThread.has(threadId)
433 ) {
434 // Startup has no publish history. Only work from this server process may
435 // produce an initial terminal alert; historical threads remain quiet.
436 if (Option.isNone(thread) || !terminalWorkSinceStart(thread.value, startedAt)) return;
437 }
420 if (publishedStateByThread.get(threadId) === publishIdentity) {438 if (publishedStateByThread.get(threadId) === publishIdentity) {
421 // The projection is back at (or never left) the last published state, so439 // The projection is back at (or never left) the last published state, so
422 // any pending deferred confirmation is moot. Leaving the deadline in440 // any pending deferred confirmation is moot. Leaving the deadline in
...@@ -495,7 +513,11 @@ export const make = Effect.gen(function* () {...@@ -495,7 +513,11 @@ export const make = Effect.gen(function* () {
495 });513 });
496 yield* Ref.update(publishedStateByThreadRef, (publishedStates) => {514 yield* Ref.update(publishedStateByThreadRef, (publishedStates) => {
497 const nextPublishedStates = new Map(publishedStates);515 const nextPublishedStates = new Map(publishedStates);
498 nextPublishedStates.set(threadId, publishIdentity);516 if (snapshot.state === null) {
517 nextPublishedStates.delete(threadId);
518 } else {
519 nextPublishedStates.set(threadId, publishIdentity);
520 }
499 return nextPublishedStates;521 return nextPublishedStates;
500 });522 });
501 });523 });
...@@ -529,6 +551,7 @@ export const make = Effect.gen(function* () {...@@ -529,6 +551,7 @@ export const make = Effect.gen(function* () {
529 const snapshot = yield* snapshotQuery.getShellSnapshot();551 const snapshot = yield* snapshotQuery.getShellSnapshot();
530 const activeThreadIds = resolveAgentAwarenessRelayActiveThreadIds({552 const activeThreadIds = resolveAgentAwarenessRelayActiveThreadIds({
531 environmentId,553 environmentId,
554 startedAt,
532 projects: snapshot.projects,555 projects: snapshot.projects,
533 threads: snapshot.threads,556 threads: snapshot.threads,
534 });557 });