Skip to content

Commit dcfb3af

Browse files
eimexdeveimexdev
committed
fix(client): advertise voice support per surface
1 parent 6be2e65 commit dcfb3af

4 files changed

Lines changed: 153 additions & 40 deletions

File tree

‎apps/web/src/connection/runtime.ts‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
import { Connection } from "@t3tools/client-runtime/connection";
22
import { shellSnapshotLoaderLayer } from "@t3tools/client-runtime/state/shell";
3-
import { threadSnapshotLoaderLayer } from "@t3tools/client-runtime/state/threads";
3+
import { makeThreadSnapshotLoaderLayer } from "@t3tools/client-runtime/state/threads";
44
import * as Layer from "effect/Layer";
55
import { Atom } from "effect/unstable/reactivity";
66

@@ -11,7 +11,10 @@ const providedConnectionPlatformLayer = connectionPlatformLayer.pipe(
1111
Layer.provide(runtimeContextLayer),
1212
);
1313

14-
const snapshotLoaderLayer = Layer.merge(threadSnapshotLoaderLayer, shellSnapshotLoaderLayer);
14+
const snapshotLoaderLayer = Layer.merge(
15+
makeThreadSnapshotLoaderLayer({ audioAttachments: true }),
16+
shellSnapshotLoaderLayer,
17+
);
1518

1619
type ConnectionLayerSource =
1720
| typeof Connection.layer

‎packages/client-runtime/src/state/threadSnapshotHttp.ts‎

Lines changed: 55 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,8 @@
1-
import type { OrchestrationThreadDetailSnapshot, ThreadId } from "@t3tools/contracts";
1+
import type {
2+
OrchestrationClientCapabilities,
3+
OrchestrationThreadDetailSnapshot,
4+
ThreadId,
5+
} from "@t3tools/contracts";
26
import * as Cause from "effect/Cause";
37
import * as Context from "effect/Context";
48
import * as Effect from "effect/Effect";
@@ -31,12 +35,14 @@ export const fetchEnvironmentThreadSnapshot = Effect.fn(
3135
)(function* (input: {
3236
readonly prepared: PreparedConnection;
3337
readonly threadId: ThreadId;
38+
readonly clientCapabilities?: OrchestrationClientCapabilities;
3439
readonly signer: Option.Option<ManagedRelayDpopSigner["Service"]>;
3540
readonly timeoutMs?: number;
3641
}) {
42+
const supportsAudioAttachments = input.clientCapabilities?.audioAttachments === true;
3743
const requestUrl = environmentEndpointUrl(
3844
input.prepared.httpBaseUrl,
39-
`/api/orchestration/threads/${input.threadId}?audioAttachments=true`,
45+
`/api/orchestration/threads/${input.threadId}${supportsAudioAttachments ? "?audioAttachments=true" : ""}`,
4046
);
4147
const client = yield* makeEnvironmentHttpApiClient(input.prepared.httpBaseUrl);
4248
const headers = yield* buildEnvironmentAuthHeaders(
@@ -52,7 +58,7 @@ export const fetchEnvironmentThreadSnapshot = Effect.fn(
5258
input.prepared.httpAuthorization,
5359
client.orchestration.threadSnapshot({
5460
params: { threadId: input.threadId },
55-
query: { audioAttachments: "true" },
61+
query: supportsAudioAttachments ? { audioAttachments: "true" } : {},
5662
headers,
5763
}),
5864
),
@@ -70,6 +76,7 @@ export type FetchEnvironmentThreadSnapshotError = RemoteEnvironmentRequestError;
7076
export class ThreadSnapshotLoader extends Context.Service<
7177
ThreadSnapshotLoader,
7278
{
79+
readonly clientCapabilities: OrchestrationClientCapabilities;
7380
readonly load: (
7481
prepared: PreparedConnection,
7582
threadId: ThreadId,
@@ -81,41 +88,53 @@ export const threadSnapshotLoaderLayer: Layer.Layer<
8188
ThreadSnapshotLoader,
8289
never,
8390
HttpClient.HttpClient
84-
> = Layer.effect(
85-
ThreadSnapshotLoader,
86-
Effect.gen(function* () {
87-
const httpClient = yield* HttpClient.HttpClient;
88-
// Resolve the DPoP signer optionally: it is only needed for relay/DPoP
89-
// connections, so the loader must not hard-require it (bearer/primary
90-
// connections work without one).
91-
const signer = yield* Effect.serviceOption(ManagedRelayDpopSigner);
92-
return ThreadSnapshotLoader.of({
93-
load: (prepared: PreparedConnection, threadId: ThreadId) =>
94-
fetchEnvironmentThreadSnapshot({ prepared, threadId, signer }).pipe(
95-
Effect.map(Option.some<OrchestrationThreadDetailSnapshot>),
96-
Effect.provideService(HttpClient.HttpClient, httpClient),
97-
// A genuinely missing thread (404) is expected — the socket
98-
// subscription is the source of truth for thread existence and will
99-
// surface the deletion — so don't treat it as an error worth warning
100-
// about; just defer to the socket path.
101-
Effect.catchTags({
102-
EnvironmentResourceNotFoundError: () =>
103-
Effect.logDebug(
104-
"Thread snapshot not found over HTTP; deferring to the socket subscription.",
91+
> = makeThreadSnapshotLoaderLayer();
92+
93+
export function makeThreadSnapshotLoaderLayer(
94+
clientCapabilities: OrchestrationClientCapabilities = {},
95+
): Layer.Layer<ThreadSnapshotLoader, never, HttpClient.HttpClient> {
96+
return Layer.effect(
97+
ThreadSnapshotLoader,
98+
Effect.gen(function* () {
99+
const httpClient = yield* HttpClient.HttpClient;
100+
// Resolve the DPoP signer optionally: it is only needed for relay/DPoP
101+
// connections, so the loader must not hard-require it (bearer/primary
102+
// connections work without one).
103+
const signer = yield* Effect.serviceOption(ManagedRelayDpopSigner);
104+
return ThreadSnapshotLoader.of({
105+
clientCapabilities,
106+
load: (prepared: PreparedConnection, threadId: ThreadId) =>
107+
fetchEnvironmentThreadSnapshot({
108+
prepared,
109+
threadId,
110+
signer,
111+
clientCapabilities,
112+
}).pipe(
113+
Effect.map(Option.some<OrchestrationThreadDetailSnapshot>),
114+
Effect.provideService(HttpClient.HttpClient, httpClient),
115+
// A genuinely missing thread (404) is expected — the socket
116+
// subscription is the source of truth for thread existence and will
117+
// surface the deletion — so don't treat it as an error worth warning
118+
// about; just defer to the socket path.
119+
Effect.catchTags({
120+
EnvironmentResourceNotFoundError: () =>
121+
Effect.logDebug(
122+
"Thread snapshot not found over HTTP; deferring to the socket subscription.",
123+
).pipe(
124+
Effect.annotateLogs({ threadId }),
125+
Effect.as(Option.none<OrchestrationThreadDetailSnapshot>()),
126+
),
127+
}),
128+
Effect.catchCause((cause) =>
129+
Effect.logWarning(
130+
"Could not load the thread snapshot over HTTP; using the socket snapshot instead.",
105131
).pipe(
106-
Effect.annotateLogs({ threadId }),
132+
Effect.annotateLogs({ threadId, cause: Cause.pretty(cause) }),
107133
Effect.as(Option.none<OrchestrationThreadDetailSnapshot>()),
108134
),
109-
}),
110-
Effect.catchCause((cause) =>
111-
Effect.logWarning(
112-
"Could not load the thread snapshot over HTTP; using the socket snapshot instead.",
113-
).pipe(
114-
Effect.annotateLogs({ threadId, cause: Cause.pretty(cause) }),
115-
Effect.as(Option.none<OrchestrationThreadDetailSnapshot>()),
116135
),
117136
),
118-
),
119-
});
120-
}),
121-
);
137+
});
138+
}),
139+
);
140+
}

‎packages/client-runtime/src/state/threads-sync.test.ts‎

Lines changed: 76 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
import {
22
EnvironmentId,
33
EventId,
4+
MessageId,
45
ORCHESTRATION_WS_METHODS,
56
ProjectId,
67
ProviderInstanceId,
@@ -133,6 +134,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o
133134
readonly cached?: OrchestrationThread;
134135
readonly httpSnapshot?: Option.Option<OrchestrationThreadDetailSnapshot>;
135136
readonly completionMarker?: boolean;
137+
readonly audioAttachments?: boolean;
136138
}) {
137139
const inputs = yield* Queue.unbounded<TestThreadInput>();
138140
const observed = yield* Queue.unbounded<EnvironmentThreadState>();
@@ -142,6 +144,9 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o
142144
const loaderCalls = yield* Ref.make(0);
143145
const lastSubscribeAfterSequence = yield* Ref.make<number | undefined>(undefined);
144146
const lastRequestCompletionMarker = yield* Ref.make<boolean | undefined>(undefined);
147+
const lastClientCapabilities = yield* Ref.make<
148+
{ readonly audioAttachments?: boolean } | undefined
149+
>(undefined);
145150
const savedThreads = yield* Ref.make<ReadonlyArray<OrchestrationThreadDetailSnapshot>>([]);
146151
const removedThreads = yield* Ref.make<ReadonlyArray<ThreadId>>([]);
147152
const wakeups = yield* Queue.unbounded<ConnectionWakeups.ConnectionWakeup>();
@@ -157,12 +162,14 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o
157162
const client = {
158163
[ORCHESTRATION_WS_METHODS.subscribeThread]: (input: {
159164
readonly afterSequence?: number;
165+
readonly clientCapabilities?: { readonly audioAttachments?: boolean };
160166
readonly requestCompletionMarker?: boolean;
161167
}) =>
162168
Stream.unwrap(
163169
Ref.updateAndGet(subscriptionCount, (count) => count + 1).pipe(
164170
Effect.andThen(Ref.set(lastSubscribeAfterSequence, input.afterSequence)),
165171
Effect.andThen(Ref.set(lastRequestCompletionMarker, input.requestCompletionMarker)),
172+
Effect.andThen(Ref.set(lastClientCapabilities, input.clientCapabilities)),
166173
Effect.as(streamFrom(inputs)),
167174
),
168175
),
@@ -179,6 +186,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o
179186
Option.some(PREPARED),
180187
);
181188
const snapshotLoader = ThreadSnapshotLoader.of({
189+
clientCapabilities: options?.audioAttachments === true ? { audioAttachments: true } : {},
182190
load: (_prepared, threadId) =>
183191
Ref.update(loaderCalls, (count) => count + 1).pipe(
184192
Effect.as(
@@ -244,6 +252,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o
244252
loaderCalls,
245253
lastSubscribeAfterSequence,
246254
lastRequestCompletionMarker,
255+
lastClientCapabilities,
247256
supervisorState,
248257
supervisorSession,
249258
savedThreads,
@@ -418,6 +427,73 @@ describe("EnvironmentThreads", () => {
418427
}),
419428
);
420429

430+
it.effect("only advertises audio attachments when the client surface opts in", () =>
431+
Effect.gen(function* () {
432+
const legacyHarness = yield* makeHarness();
433+
yield* Queue.offer(legacyHarness.inputs, snapshot(BASE_THREAD));
434+
yield* awaitThreadState(legacyHarness.observed, (value) => value.status === "live");
435+
expect(yield* Ref.get(legacyHarness.lastClientCapabilities)).toBeUndefined();
436+
437+
const capableHarness = yield* makeHarness({ audioAttachments: true });
438+
yield* Queue.offer(capableHarness.inputs, snapshot(BASE_THREAD));
439+
yield* awaitThreadState(capableHarness.observed, (value) => value.status === "live");
440+
expect(yield* Ref.get(capableHarness.lastClientCapabilities)).toEqual({
441+
audioAttachments: true,
442+
});
443+
}),
444+
);
445+
446+
it.effect("refreshes a cached audio message for a legacy client", () =>
447+
Effect.gen(function* () {
448+
const cachedThread: OrchestrationThread = {
449+
...BASE_THREAD,
450+
messages: [
451+
{
452+
id: MessageId.make("voice-message"),
453+
role: "user",
454+
text: "",
455+
attachments: [
456+
{
457+
type: "audio",
458+
id: "audio-1",
459+
name: "Voice note",
460+
mimeType: "audio/webm",
461+
sizeBytes: 100,
462+
durationMs: 1_000,
463+
waveform: [0.5],
464+
},
465+
],
466+
turnId: null,
467+
streaming: false,
468+
createdAt: BASE_THREAD.createdAt,
469+
updatedAt: BASE_THREAD.createdAt,
470+
},
471+
],
472+
};
473+
const projectedThread: OrchestrationThread = {
474+
...cachedThread,
475+
messages: [{ ...cachedThread.messages[0]!, text: "Voice transcript", attachments: [] }],
476+
};
477+
const harness = yield* makeHarness({
478+
cached: cachedThread,
479+
httpSnapshot: Option.some({
480+
snapshotSequence: CACHED_SNAPSHOT_SEQUENCE + 1,
481+
thread: projectedThread,
482+
}),
483+
});
484+
485+
const state = yield* awaitThreadState(
486+
harness.observed,
487+
(value) =>
488+
Option.isSome(value.data) && value.data.value.messages[0]?.text === "Voice transcript",
489+
);
490+
491+
expect(Option.getOrThrow(state.data).messages[0]?.attachments).toEqual([]);
492+
expect(yield* Ref.get(harness.loaderCalls)).toBeGreaterThanOrEqual(1);
493+
expect(yield* Ref.get(harness.lastSubscribeAfterSequence)).toBe(CACHED_SNAPSHOT_SEQUENCE + 1);
494+
}),
495+
);
496+
421497
it.effect("ignores replayed thread events at or below the snapshot sequence", () =>
422498
Effect.gen(function* () {
423499
const harness = yield* makeHarness({ cached: BASE_THREAD });

‎packages/client-runtime/src/state/threads.ts‎

Lines changed: 17 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,12 @@ function shouldPersistThread(thread: OrchestrationThread): boolean {
4848
return status !== "starting" && status !== "running";
4949
}
5050

51+
function containsAudioAttachment(thread: OrchestrationThread): boolean {
52+
return thread.messages.some((message) =>
53+
message.attachments?.some((attachment) => attachment.type === "audio"),
54+
);
55+
}
56+
5157
export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make")(function* (
5258
threadId: ThreadIdType,
5359
) {
@@ -252,7 +258,14 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
252258
yield* setSynchronizing;
253259

254260
let current = yield* SubscriptionRef.get(state);
255-
if (Option.isNone(current.data) && current.status !== "deleted") {
261+
const cachedAudioNeedsCompatibilityRefresh =
262+
snapshotLoader.clientCapabilities.audioAttachments !== true &&
263+
Option.isSome(current.data) &&
264+
containsAudioAttachment(current.data.value);
265+
if (
266+
(Option.isNone(current.data) || cachedAudioNeedsCompatibilityRefresh) &&
267+
current.status !== "deleted"
268+
) {
256269
const prepared = yield* SubscriptionRef.get(supervisor.prepared).pipe(
257270
Effect.flatMap(
258271
Option.match({
@@ -286,7 +299,9 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
286299

287300
return {
288301
threadId,
289-
clientCapabilities: { audioAttachments: true as const },
302+
...(snapshotLoader.clientCapabilities.audioAttachments === true
303+
? { clientCapabilities: { audioAttachments: true as const } }
304+
: {}),
290305
...(canResume ? { afterSequence: sequence } : {}),
291306
...(supportsCompletionMarker ? { requestCompletionMarker: true as const } : {}),
292307
};

0 commit comments

Comments
 (0)