authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-03-28 10:51:01-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-03-28 10:51:01-07:00
log3e2df5a7823d180f634370998d58497993b94ee7
tree4b49a45007e9438bb6600d37a97d958de39d5cdd
parentf0cdaf94c2b69a0f6a55f7ca2e3d13782b4272a7
signature Signed by PGP key B5690EEEBB952194

Normalize typed provider runtime ingestion (#1475)

2 files changed, 73 insertions(+), 41 deletions(-)

apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts+65-3
...@@ -4,7 +4,6 @@ import path from "node:path";...@@ -4,7 +4,6 @@ import path from "node:path";
44
5import type {5import type {
6 OrchestrationReadModel,6 OrchestrationReadModel,
7 ProviderKind,
8 ProviderRuntimeEvent,7 ProviderRuntimeEvent,
9 ProviderSession,8 ProviderSession,
10} from "@t3tools/contracts";9} from "@t3tools/contracts";
...@@ -56,7 +55,7 @@ const asTurnId = (value: string): TurnId => TurnId.makeUnsafe(value);...@@ -56,7 +55,7 @@ const asTurnId = (value: string): TurnId => TurnId.makeUnsafe(value);
56type LegacyProviderRuntimeEvent = {55type LegacyProviderRuntimeEvent = {
57 readonly type: string;56 readonly type: string;
58 readonly eventId: EventId;57 readonly eventId: EventId;
59 readonly provider: ProviderKind;58 readonly provider: ProviderRuntimeEvent["provider"];
60 readonly createdAt: string;59 readonly createdAt: string;
61 readonly threadId: ThreadId;60 readonly threadId: ThreadId;
62 readonly turnId?: string | undefined;61 readonly turnId?: string | undefined;
...@@ -66,6 +65,23 @@ type LegacyProviderRuntimeEvent = {...@@ -66,6 +65,23 @@ type LegacyProviderRuntimeEvent = {
66 readonly [key: string]: unknown;65 readonly [key: string]: unknown;
67};66};
6867
68type LegacyTurnCompletedEvent = LegacyProviderRuntimeEvent & {
69 readonly type: "turn.completed";
70 readonly payload?: undefined;
71 readonly status: "completed" | "failed" | "interrupted" | "cancelled";
72 readonly errorMessage?: string | undefined;
73};
74
75function isLegacyTurnCompletedEvent(
76 event: LegacyProviderRuntimeEvent,
77): event is LegacyTurnCompletedEvent {
78 return (
79 event.type === "turn.completed" &&
80 event.payload === undefined &&
81 typeof event.status === "string"
82 );
83}
84
69function createProviderServiceHarness() {85function createProviderServiceHarness() {
70 const runtimeEventPubSub = Effect.runSync(PubSub.unbounded<ProviderRuntimeEvent>());86 const runtimeEventPubSub = Effect.runSync(PubSub.unbounded<ProviderRuntimeEvent>());
71 const runtimeSessions: ProviderSession[] = [];87 const runtimeSessions: ProviderSession[] = [];
...@@ -93,8 +109,23 @@ function createProviderServiceHarness() {...@@ -93,8 +109,23 @@ function createProviderServiceHarness() {
93 runtimeSessions.push(session);109 runtimeSessions.push(session);
94 };110 };
95111
112 const normalizeLegacyEvent = (event: LegacyProviderRuntimeEvent): ProviderRuntimeEvent => {
113 if (isLegacyTurnCompletedEvent(event)) {
114 const normalized: Extract<ProviderRuntimeEvent, { type: "turn.completed" }> = {
115 ...(event as Omit<Extract<ProviderRuntimeEvent, { type: "turn.completed" }>, "payload">),
116 payload: {
117 state: event.status,
118 ...(typeof event.errorMessage === "string" ? { errorMessage: event.errorMessage } : {}),
119 },
120 };
121 return normalized;
122 }
123
124 return event as ProviderRuntimeEvent;
125 };
126
96 const emit = (event: LegacyProviderRuntimeEvent): void => {127 const emit = (event: LegacyProviderRuntimeEvent): void => {
97 Effect.runSync(PubSub.publish(runtimeEventPubSub, event as unknown as ProviderRuntimeEvent));128 Effect.runSync(PubSub.publish(runtimeEventPubSub, normalizeLegacyEvent(event)));
98 };129 };
99130
100 return {131 return {
...@@ -1695,6 +1726,37 @@ describe("ProviderRuntimeIngestion", () => {...@@ -1695,6 +1726,37 @@ describe("ProviderRuntimeIngestion", () => {
1695 expect(thread.session?.lastError).toBe("runtime exploded");1726 expect(thread.session?.lastError).toBe("runtime exploded");
1696 });1727 });
16971728
1729 it("records runtime.error activities from the typed payload message", async () => {
1730 const harness = await createHarness();
1731 const now = new Date().toISOString();
1732
1733 harness.emit({
1734 type: "runtime.error",
1735 eventId: asEventId("evt-runtime-error-activity"),
1736 provider: "codex",
1737 createdAt: now,
1738 threadId: asThreadId("thread-1"),
1739 turnId: asTurnId("turn-runtime-error-activity"),
1740 payload: {
1741 message: "runtime activity exploded",
1742 },
1743 });
1744
1745 const thread = await waitForThread(harness.engine, (entry) =>
1746 entry.activities.some((activity) => activity.id === "evt-runtime-error-activity"),
1747 );
1748 const activity = thread.activities.find(
1749 (entry: ProviderRuntimeTestActivity) => entry.id === "evt-runtime-error-activity",
1750 );
1751 const activityPayload =
1752 activity?.payload && typeof activity.payload === "object"
1753 ? (activity.payload as Record<string, unknown>)
1754 : undefined;
1755
1756 expect(activity?.kind).toBe("runtime.error");
1757 expect(activityPayload?.message).toBe("runtime activity exploded");
1758 });
1759
1698 it("keeps the session running when a runtime.warning arrives during an active turn", async () => {1760 it("keeps the session running when a runtime.warning arrives during an active turn", async () => {
1699 const harness = await createHarness();1761 const harness = await createHarness();
1700 const now = new Date().toISOString();1762 const now = new Date().toISOString();
apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts+8-38
...@@ -98,10 +98,6 @@ function proposedPlanIdFromEvent(event: ProviderRuntimeEvent, threadId: ThreadId...@@ -98,10 +98,6 @@ function proposedPlanIdFromEvent(event: ProviderRuntimeEvent, threadId: ThreadId
98 return `plan:${threadId}:event:${event.eventId}`;98 return `plan:${threadId}:event:${event.eventId}`;
99}99}
100100
101function asString(value: unknown): string | undefined {
102 return typeof value === "string" ? value : undefined;
103}
104
105function buildContextWindowActivityPayload(101function buildContextWindowActivityPayload(
106 event: ProviderRuntimeEvent,102 event: ProviderRuntimeEvent,
107): ThreadTokenUsageSnapshot | undefined {103): ThreadTokenUsageSnapshot | undefined {
...@@ -111,14 +107,6 @@ function buildContextWindowActivityPayload(...@@ -111,14 +107,6 @@ function buildContextWindowActivityPayload(
111 return event.payload.usage;107 return event.payload.usage;
112}108}
113109
114function runtimePayloadRecord(event: ProviderRuntimeEvent): Record<string, unknown> | undefined {
115 const payload = (event as { payload?: unknown }).payload;
116 if (!payload || typeof payload !== "object") {
117 return undefined;
118 }
119 return payload as Record<string, unknown>;
120}
121
122function normalizeRuntimeTurnState(110function normalizeRuntimeTurnState(
123 value: string | undefined,111 value: string | undefined,
124): "completed" | "failed" | "interrupted" | "cancelled" {112): "completed" | "failed" | "interrupted" | "cancelled" {
...@@ -133,23 +121,6 @@ function normalizeRuntimeTurnState(...@@ -133,23 +121,6 @@ function normalizeRuntimeTurnState(
133 }121 }
134}122}
135123
136function runtimeTurnState(
137 event: ProviderRuntimeEvent,
138): "completed" | "failed" | "interrupted" | "cancelled" {
139 const payloadState = asString(runtimePayloadRecord(event)?.state);
140 return normalizeRuntimeTurnState(payloadState);
141}
142
143function runtimeTurnErrorMessage(event: ProviderRuntimeEvent): string | undefined {
144 const payloadErrorMessage = asString(runtimePayloadRecord(event)?.errorMessage);
145 return payloadErrorMessage;
146}
147
148function runtimeErrorMessageFromEvent(event: ProviderRuntimeEvent): string | undefined {
149 const payloadMessage = asString(runtimePayloadRecord(event)?.message);
150 return payloadMessage;
151}
152
153function orchestrationSessionStatusFromRuntimeState(124function orchestrationSessionStatusFromRuntimeState(
154 state: "starting" | "running" | "waiting" | "ready" | "interrupted" | "stopped" | "error",125 state: "starting" | "running" | "waiting" | "ready" | "interrupted" | "stopped" | "error",
155): "starting" | "running" | "ready" | "interrupted" | "stopped" | "error" {126): "starting" | "running" | "ready" | "interrupted" | "stopped" | "error" {
...@@ -253,10 +224,6 @@ function runtimeEventToActivities(...@@ -253,10 +224,6 @@ function runtimeEventToActivities(
253 }224 }
254225
255 case "runtime.error": {226 case "runtime.error": {
256 const message = runtimeErrorMessageFromEvent(event);
257 if (!message) {
258 return [];
259 }
260 return [227 return [
261 {228 {
262 id: event.eventId,229 id: event.eventId,
...@@ -265,7 +232,7 @@ function runtimeEventToActivities(...@@ -265,7 +232,7 @@ function runtimeEventToActivities(
265 kind: "runtime.error",232 kind: "runtime.error",
266 summary: "Runtime error",233 summary: "Runtime error",
267 payload: {234 payload: {
268 message: truncateDetail(message),235 message: truncateDetail(event.payload.message),
269 },236 },
270 turnId: toTurnId(event.turnId) ?? null,237 turnId: toTurnId(event.turnId) ?? null,
271 ...maybeSequence,238 ...maybeSequence,
...@@ -973,7 +940,9 @@ const make = Effect.gen(function* () {...@@ -973,7 +940,9 @@ const make = Effect.gen(function* () {
973 case "session.exited":940 case "session.exited":
974 return "stopped";941 return "stopped";
975 case "turn.completed":942 case "turn.completed":
976 return runtimeTurnState(event) === "failed" ? "error" : "ready";943 return normalizeRuntimeTurnState(event.payload.state) === "failed"
944 ? "error"
945 : "ready";
977 case "session.started":946 case "session.started":
978 case "thread.started":947 case "thread.started":
979 // Provider thread/session start notifications can arrive during an948 // Provider thread/session start notifications can arrive during an
...@@ -984,8 +953,9 @@ const make = Effect.gen(function* () {...@@ -984,8 +953,9 @@ const make = Effect.gen(function* () {
984 const lastError =953 const lastError =
985 event.type === "session.state.changed" && event.payload.state === "error"954 event.type === "session.state.changed" && event.payload.state === "error"
986 ? (event.payload.reason ?? thread.session?.lastError ?? "Provider session error")955 ? (event.payload.reason ?? thread.session?.lastError ?? "Provider session error")
987 : event.type === "turn.completed" && runtimeTurnState(event) === "failed"956 : event.type === "turn.completed" &&
988 ? (runtimeTurnErrorMessage(event) ?? thread.session?.lastError ?? "Turn failed")957 normalizeRuntimeTurnState(event.payload.state) === "failed"
958 ? (event.payload.errorMessage ?? thread.session?.lastError ?? "Turn failed")
989 : status === "ready"959 : status === "ready"
990 ? null960 ? null
991 : (thread.session?.lastError ?? null);961 : (thread.session?.lastError ?? null);
...@@ -1176,7 +1146,7 @@ const make = Effect.gen(function* () {...@@ -1176,7 +1146,7 @@ const make = Effect.gen(function* () {
1176 }1146 }
11771147
1178 if (event.type === "runtime.error") {1148 if (event.type === "runtime.error") {
1179 const runtimeErrorMessage = runtimeErrorMessageFromEvent(event) ?? "Provider runtime error";1149 const runtimeErrorMessage = event.payload.message;
11801150
1181 const shouldApplyRuntimeError = !STRICT_PROVIDER_LIFECYCLE_GUARD1151 const shouldApplyRuntimeError = !STRICT_PROVIDER_LIFECYCLE_GUARD
1182 ? true1152 ? true