k1nq/TREK
1
0
Fork 0
mirror of https://github.com/mauriceboe/TREK.git synced 2026-10-23 23:46:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(assistant): lint, types and tests for the CI round

Browse Source 
- agent loop and decisions without await in loops: steps as recursion,
  calls and applied changes in order through one promise chain, the
  independent reads and claims in parallel; part mapping in its own helpers
- place card, header and weather strip without refs or components created
  in render; long hooks and components split; state reset during render
- trip notes only in a turn's tool session, never an undo's; map view
  reported only while the assistant can run
- place strip scrolls through the strip's own smooth scroll, no physical side
- tests follow the raised limits and the new card fields; toasts read through
  a mocked useToast instead of a window global; baselines lowered where the
  counts fell
This commit is contained in:
Maurice
2026-10-21 21:45:25 +02:00
parent b6f7488591
commit 3c3a228737
47 changed files with 892 additions and 610 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/src/nest/assistant/assistant-agent.ts
+237 -168
View File
@@ -172,6 +172,231 @@ function uniqueCallId(call: AiToolCall, used: Set<string>, newId: () => string):
return id;
}
/** What every step of one turn shares. */
interface Turn {
deps: AgentDeps;
input: AgentInput;
w: UiStreamWriter;
signal: AbortSignal;
history: AiChatMessage[];
usage: { inputTokens: number; outputTokens: number };
knownPlaceIds: Set<string>;
/** Whether the turn used its one push (an empty, cut-off or announcing answer). */
nudged: boolean;
}
function outcomeOf(turn: Turn, end: AgentOutcome['end'], code: AssistantErrorCode | null = null): AgentOutcome {
return { end, code, ...turn.usage };
}
function finishTurn(turn: Turn, reason: 'stop' | 'tool-calls' | 'error'): void {
if (turn.deps.quota) turn.w.quota(turn.deps.quota.used, turn.deps.quota.limit);
turn.w.finish(reason);
}
/** What the model answered in one step. */
interface StepAnswer {
text: string;
received: AiToolCall[];
providerState: AiProviderState | null;
cutOff: boolean;
}
/** Streams one answer of the model; a failure or a stop ends the turn here, saved as far as it got. */
async function streamAnswer(turn: Turn, rowId: number): Promise<StepAnswer | AgentOutcome> {
const { deps, input, w, signal } = turn;
const answer: StepAnswer = { text: '', received: [], providerState: null, cutOff: false };
try {
const events = deps.client.stream(
{
system: input.system,
context: input.context,
messages: turn.history,
tools: deps.tools.specs(),
cacheKey: input.cacheKey,
},
signal,
);
for await (const event of events) {
if (event.type === 'text-delta') {
answer.text += event.delta;
w.text(event.delta);
} else if (event.type === 'tool-call') answer.received.push(event.call);
else if (event.type === 'usage') {
turn.usage.inputTokens += event.inputTokens;
turn.usage.outputTokens += event.outputTokens;
} else if (event.type === 'provider-state') answer.providerState = event.state;
else if (event.type === 'finish') answer.cutOff = event.reason === 'length';
}
return answer;
} catch (err) {
const partial: StepRecord = {
parts: answer.text ? [{ type: 'text', text: answer.text }] : [],
providerState: null,
};
if (signal.aborted) {
await deps.store.saveStep(rowId, partial, 'aborted');
w.abort();
return outcomeOf(turn, 'aborted');
}
const code = err instanceof AiProviderError ? err.code : 'generic';
await deps.store.saveStep(rowId, partial, 'error');
w.error(code);
finishTurn(turn, 'error');
return outcomeOf(turn, 'failed', code);
}
}
/**
* Runs the calls of one step in the order the model made them: a read at once,
* a change described and held, a refused call answered with why.
*/
async function runCalls(turn: Turn, calls: AiToolCall[], collected: StepParts): Promise<HeldCall[]> {
const { deps, w, signal } = turn;
const writes = calls.filter((call) => deps.tools.kindOf(call.name) === 'write').length;
const held: HeldCall[] = [];
let reads = 0;
const runOne = async (call: AiToolCall): Promise<void> => {
if (signal.aborted) return;
const kind = deps.tools.kindOf(call.name);
let prepared = kind ? deps.tools.pinArgs(call.name, call.args) : { error: `Unknown tool: ${call.name}` };
const unknownPlace = 'error' in prepared ? null : placeIdRefusal(call.name, prepared.args, turn.knownPlaceIds);
if (unknownPlace) prepared = { error: unknownPlace };
if (kind === 'read' && !('error' in prepared)) reads += 1;
const overCap = kind === 'write' ? writes > MAX_WRITES_PER_STEP : reads > MAX_READS_PER_STEP;
if ('error' in prepared || overCap) {
let refusal = kind === 'write' ? TOO_MANY_CHANGES : TOO_MANY_LOOKUPS;
if ('error' in prepared) refusal = prepared.error;
w.toolInput(call.id, call.name, call.args);
w.toolError(call.id, refusal);
collected.parts.push({
type: 'tool-call',
toolCallId: call.id,
toolName: call.name,
args: call.args,
result: refusal,
isError: true,
});
return;
}
if (kind === 'read') {
await runRead(deps, { id: call.id, name: call.name, args: prepared.args }, w, collected);
return;
}
const id = deps.newId();
const preview = await deps.describe(call.name, id, prepared.args);
held.push({ id, providerCallId: call.id, toolName: call.name, args: prepared.args, preview });
};
// One after another: the stream tells each call before the next one starts.
await calls.reduce<Promise<void>>(async (previous, call) => {
await previous;
await runOne(call);
}, Promise.resolve());
return held;
}
/** Saves a step whose changes wait for the user and announces them; the turn ends there. */
async function holdChanges(turn: Turn, rowId: number, step: StepRecord, held: HeldCall[]): Promise<AgentOutcome> {
const { w } = turn;
for (const change of held) {
const { id, toolName, args } = change;
step.parts.push({ type: 'tool-call', toolCallId: id, toolName, args, approval: { id } });
}
step.parts.push(...held.map((change): AssistantMessagePart => ({ type: 'data-trek-change', data: change.preview })));
await turn.deps.store.holdStep(rowId, step, held);
for (const change of held) {
w.toolInput(change.id, change.toolName, change.args);
w.approvalRequest(change.id);
w.change(change.preview);
}
w.finishStep();
finishTurn(turn, 'tool-calls');
return outcomeOf(turn, 'awaiting');
}
/** Records a step that ran its reads, and hands their answers back for the next one. */
async function carryOn(turn: Turn, rowId: number, step: StepRecord, text: string, calls: AiToolCall[]): Promise<void> {
await turn.deps.store.saveStep(rowId, step, 'complete');
turn.w.finishStep();
turn.history.push({ role: 'assistant', text, toolCalls: calls, providerState: step.providerState });
for (const part of step.parts) {
if (part.type !== 'tool-call') continue;
turn.history.push({
role: 'tool',
toolCallId: part.toolCallId,
name: part.toolName,
output: outputText(part.result),
...(part.isError ? { isError: true } : {}),
});
}
}
/** One step of the turn: the outcome when the turn ends with it, null when the model is asked again. */
async function runStep(turn: Turn, step: number): Promise<AgentOutcome | null> {
const { deps, input, w, signal, history } = turn;
const rowId = await deps.store.openStep();
w.startStep();
const answer = await streamAnswer(turn, rowId);
if ('end' in answer) return answer;
const { text, received, providerState, cutOff } = answer;
const parts: AssistantMessagePart[] = text ? [{ type: 'text', text }] : [];
const empty = received.length === 0 && !text.trim();
const announced = received.length === 0 && step === 0 && ANNOUNCED.test(text.trimEnd());
const push = empty || (received.length === 0 && cutOff) ? GO_ON : announced ? MAKE_THE_CALLS : null;
if (push && !turn.nudged) {
turn.nudged = true;
await deps.store.saveStep(rowId, { parts, providerState }, 'complete');
w.finishStep();
// An empty assistant message is one the providers refuse; only the push goes back.
if (!empty) history.push({ role: 'assistant', text, toolCalls: [], providerState });
history.push({ role: 'user', text: push });
return null;
}
if (empty) {
await deps.store.saveStep(rowId, { parts, providerState }, 'error');
w.error('empty_answer');
finishTurn(turn, 'error');
return outcomeOf(turn, 'failed', 'empty_answer');
}
if (received.length === 0) {
await deps.store.saveStep(rowId, { parts, providerState }, 'complete');
w.finishStep();
finishTurn(turn, 'stop');
return outcomeOf(turn, 'stop');
}
const calls = received.map((call) => ({ ...call, id: uniqueCallId(call, input.usedCallIds, deps.newId) }));
const collected: StepParts = { parts, weather: [], knownPlaceIds: turn.knownPlaceIds };
const held = await runCalls(turn, calls, collected);
// One weather strip per step, whatever the step held: it is told before the changes.
const weather = weatherOf(collected.weather);
if (weather) {
w.weather(weather);
parts.push({ type: 'data-trek-weather', data: weather });
}
// A stopped step keeps what ran; the changes it was about to propose are dropped, never held.
if (signal.aborted) {
await deps.store.saveStep(rowId, { parts, providerState }, 'aborted');
w.abort();
return outcomeOf(turn, 'aborted');
}
if (held.length > 0) return holdChanges(turn, rowId, { parts, providerState }, held);
await carryOn(turn, rowId, { parts, providerState }, text, calls);
return null;
}
/** The steps from `step` on, each after the one before, until one ends the turn or the limit does. */
async function stepsFrom(turn: Turn, step: number): Promise<AgentOutcome> {
if (step >= MAX_STEPS) {
turn.w.error('max_steps');
finishTurn(turn, 'error');
return outcomeOf(turn, 'max_steps', 'max_steps');
}
return (await runStep(turn, step)) ?? stepsFrom(turn, step + 1);
}
/**
* One assistant turn: the model is asked, the reads it calls are run and
* answered, and it is asked again, until it answers without calling a tool,
@@ -183,180 +408,24 @@ function uniqueCallId(call: AiToolCall, used: Set<string>, newId: () => string):
* before its held changes are announced, so a card the client shows always
* exists on the server.
*/
export async function runAgentLoop(
export function runAgentLoop(
deps: AgentDeps,
input: AgentInput,
w: UiStreamWriter,
signal: AbortSignal,
): Promise<AgentOutcome> {
const usage = { inputTokens: 0, outputTokens: 0 };
const history = [...input.history];
const outcome = (end: AgentOutcome['end'], code: AssistantErrorCode | null = null): AgentOutcome => ({
end,
code,
...usage,
});
const finish = (reason: 'stop' | 'tool-calls' | 'error') => {
if (deps.quota) w.quota(deps.quota.used, deps.quota.limit);
w.finish(reason);
};
const knownPlaceIds = new Set(
[...deps.card.places.values()].flatMap((place) => (place.googlePlaceId ? [place.googlePlaceId] : [])),
);
let nudged = false;
for (let step = 0; step < MAX_STEPS; step++) {
const rowId = await deps.store.openStep();
w.startStep();
let text = '';
let providerState: AiProviderState | null = null;
let cutOff = false;
const received: AiToolCall[] = [];
try {
const events = deps.client.stream(
{
system: input.system,
context: input.context,
messages: history,
tools: deps.tools.specs(),
cacheKey: input.cacheKey,
},
signal,
);
for await (const event of events) {
if (event.type === 'text-delta') {
text += event.delta;
w.text(event.delta);
} else if (event.type === 'tool-call') received.push(event.call);
else if (event.type === 'usage') {
usage.inputTokens += event.inputTokens;
usage.outputTokens += event.outputTokens;
} else if (event.type === 'provider-state') providerState = event.state;
else if (event.type === 'finish') cutOff = event.reason === 'length';
}
} catch (err) {
const partial: StepRecord = { parts: text ? [{ type: 'text', text }] : [], providerState: null };
if (signal.aborted) {
await deps.store.saveStep(rowId, partial, 'aborted');
w.abort();
return outcome('aborted');
}
const code = err instanceof AiProviderError ? err.code : 'generic';
await deps.store.saveStep(rowId, partial, 'error');
w.error(code);
finish('error');
return outcome('failed', code);
}
const parts: AssistantMessagePart[] = text ? [{ type: 'text', text }] : [];
const empty = received.length === 0 && !text.trim();
const announced = received.length === 0 && step === 0 && ANNOUNCED.test(text.trimEnd());
const push = empty || (received.length === 0 && cutOff) ? GO_ON : announced ? MAKE_THE_CALLS : null;
if (push && !nudged) {
nudged = true;
await deps.store.saveStep(rowId, { parts, providerState }, 'complete');
w.finishStep();
// An empty assistant message is one the providers refuse; only the push goes back.
if (!empty) history.push({ role: 'assistant', text, toolCalls: [], providerState });
history.push({ role: 'user', text: push });
continue;
}
if (empty) {
await deps.store.saveStep(rowId, { parts, providerState }, 'error');
w.error('empty_answer');
finish('error');
return outcome('failed', 'empty_answer');
}
if (received.length === 0) {
await deps.store.saveStep(rowId, { parts, providerState }, 'complete');
w.finishStep();
finish('stop');
return outcome('stop');
}
const calls = received.map((call) => ({ ...call, id: uniqueCallId(call, input.usedCallIds, deps.newId) }));
const writes = calls.filter((call) => deps.tools.kindOf(call.name) === 'write').length;
const held: HeldCall[] = [];
const collected: StepParts = { parts, weather: [], knownPlaceIds };
let reads = 0;
for (const call of calls) {
if (signal.aborted) break;
const kind = deps.tools.kindOf(call.name);
let prepared = kind ? deps.tools.pinArgs(call.name, call.args) : { error: `Unknown tool: ${call.name}` };
const unknownPlace = 'error' in prepared ? null : placeIdRefusal(call.name, prepared.args, knownPlaceIds);
if (unknownPlace) prepared = { error: unknownPlace };
if (kind === 'read' && !('error' in prepared)) reads += 1;
const overCap = kind === 'write' ? writes > MAX_WRITES_PER_STEP : reads > MAX_READS_PER_STEP;
if ('error' in prepared || overCap) {
let refusal = kind === 'write' ? TOO_MANY_CHANGES : TOO_MANY_LOOKUPS;
if ('error' in prepared) refusal = prepared.error;
w.toolInput(call.id, call.name, call.args);
w.toolError(call.id, refusal);
parts.push({
type: 'tool-call',
toolCallId: call.id,
toolName: call.name,
args: call.args,
result: refusal,
isError: true,
});
continue;
}
if (kind === 'read') {
await runRead(deps, { id: call.id, name: call.name, args: prepared.args }, w, collected);
continue;
}
const id = deps.newId();
const preview = await deps.describe(call.name, id, prepared.args);
held.push({ id, providerCallId: call.id, toolName: call.name, args: prepared.args, preview });
}
// One weather strip per step, whatever the step held: it is told before the changes.
const weather = weatherOf(collected.weather);
if (weather) {
w.weather(weather);
parts.push({ type: 'data-trek-weather', data: weather });
}
// A stopped step keeps what ran; the changes it was about to propose are dropped, never held.
if (signal.aborted) {
await deps.store.saveStep(rowId, { parts, providerState }, 'aborted');
w.abort();
return outcome('aborted');
}
if (held.length > 0) {
for (const change of held) {
const { id, toolName, args } = change;
parts.push({ type: 'tool-call', toolCallId: id, toolName, args, approval: { id } });
}
parts.push(...held.map((change): AssistantMessagePart => ({ type: 'data-trek-change', data: change.preview })));
await deps.store.holdStep(rowId, { parts, providerState }, held);
for (const change of held) {
w.toolInput(change.id, change.toolName, change.args);
w.approvalRequest(change.id);
w.change(change.preview);
}
w.finishStep();
finish('tool-calls');
return outcome('awaiting');
}
await deps.store.saveStep(rowId, { parts, providerState }, 'complete');
w.finishStep();
history.push({ role: 'assistant', text, toolCalls: calls, providerState });
for (const part of parts) {
if (part.type !== 'tool-call') continue;
history.push({
role: 'tool',
toolCallId: part.toolCallId,
name: part.toolName,
output: outputText(part.result),
...(part.isError ? { isError: true } : {}),
});
}
}
w.error('max_steps');
finish('error');
return outcome('max_steps', 'max_steps');
const turn: Turn = {
deps,
input,
w,
signal,
history: [...input.history],
usage: { inputTokens: 0, outputTokens: 0 },
knownPlaceIds,
nudged: false,
};
return stepsFrom(turn, 0);
}
server/src/nest/assistant/assistant-cards.helpers.ts
+9 -5
View File
@@ -1,6 +1,6 @@
import type { CardPlace, TripCard } from './assistant-context';
import { redactForProvider } from './assistant-redact';
import { dayLabel } from './tool-call-preview.helpers';
import { dayLabel, type PreviewContext } from './tool-call-preview.helpers';
import {
haversineMetres,
kindOfResult,
@@ -255,6 +255,13 @@ export function activityOf(toolName: string, toolCallId: string, args: Row, card
return label === null && area === null ? null : { toolCallId, label, area };
}
/** What a mention is called in the trip; null when the trip holds no such thing. */
function mentionLabel(ctx: PreviewContext, kind: AssistantMention['kind'], id: number): string | null {
if (kind === 'day') return ctx.card.days.some((day) => day.id === id) ? dayLabel(ctx, id) : null;
if (kind === 'place') return ctx.card.places.get(id)?.name ?? null;
return ctx.card.reservations.get(id)?.title ?? null;
}
/**
* The days, places and bookings the user named with `@`, with the labels the
* trip gives them (never the ones the request sent). One that is not in the
@@ -269,10 +276,7 @@ export function mentionItems(
const seen = new Set<string>();
const ctx = { card, categories: new Map<number, string>(), locale };
for (const { kind, id } of mentions) {
let label: string | null = null;
if (kind === 'day') label = card.days.some((day) => day.id === id) ? dayLabel(ctx, id) : null;
else if (kind === 'place') label = card.places.get(id)?.name ?? null;
else label = card.reservations.get(id)?.title ?? null;
const label = mentionLabel(ctx, kind, id);
const shown = clipped(label, MAX_NAME_CHARS);
if (!shown || seen.has(`${kind}:${id}`)) continue;
seen.add(`${kind}:${id}`);
server/src/nest/assistant/assistant-decisions.helpers.ts
+44
View File
@@ -0,0 +1,44 @@
import type { AssistantMessagePart } from '@trek/shared';
/** What the model is told about a change the user discarded. */
export const DISCARDED = { applied: false, reason: 'The user discarded this change.' };
/** ... one the user left undecided by asking something else. */
export const CANCELLED = { applied: false, reason: 'The user moved on without deciding.' };
/** ... one that was approved but never ran. */
export const NOT_RUN = { applied: false, reason: 'The change could not be run.' };
/** What became of one decided change, as its message records it. */
export interface Outcome {
approved: boolean;
result: unknown;
isError: boolean;
}
/** The parts of a message with the named held changes marked as left undecided. */
export function withCancelled(
parts: readonly AssistantMessagePart[],
cancelled: ReadonlySet<string>,
): AssistantMessagePart[] {
return parts.map((part) =>
part.type === 'tool-call' && cancelled.has(part.toolCallId)
? { ...part, result: CANCELLED, approval: { id: part.toolCallId, resolution: 'cancelled' as const } }
: part,
);
}
/** The parts of a message with each decided change carrying its outcome. */
export function withOutcomes(
parts: readonly AssistantMessagePart[],
outcomes: ReadonlyMap<string, Outcome>,
): AssistantMessagePart[] {
return parts.map((part) => {
const outcome = part.type === 'tool-call' ? outcomes.get(part.toolCallId) : undefined;
if (part.type !== 'tool-call' || !outcome) return part;
return {
...part,
result: outcome.result,
...(outcome.isError ? { isError: true } : {}),
approval: { id: part.toolCallId, approved: outcome.approved },
};
});
}
server/src/nest/assistant/assistant-found-places.service.ts
+3 -4
View File
@@ -213,10 +213,9 @@ export class AssistantFoundPlacesService {
private async answerOf(threadId: number): Promise<AssistantMessage> {
const rows = await this.messages.listLatest(threadId, THREAD_ROWS);
const calls = new Map<string, AssistantToolCallRow>();
for (const row of rows) {
if (!partsOf(row).some((part) => part.type === 'tool-call' && part.approval)) continue;
for (const call of await this.toolCalls.listByMessage(row.id)) calls.set(call.id, call);
}
const held = rows.filter((row) => partsOf(row).some((part) => part.type === 'tool-call' && part.approval));
const lists = await Promise.all(held.map((row) => this.toolCalls.listByMessage(row.id)));
for (const call of lists.flat()) calls.set(call.id, call);
const messages = toThreadMessages(rows, calls, new Date());
return messages[messages.length - 1];
}
server/src/nest/assistant/assistant-tools.service.ts
+6 -3
View File
@@ -127,6 +127,8 @@ export class AssistantToolSession {
private readonly tools: ReadonlyMap<string, ListedTool>,
private readonly tripId: number,
private readonly audit: (toolName: string) => void,
/** Whether the model may keep trip notes here: a turn's session, never an undo's. */
private readonly offersNotes: boolean,
) {}
/** The tools as the model is offered them, with the trip notes the server keeps itself. */
@@ -136,11 +138,11 @@ export class AssistantToolSession {
description: tool.description,
parameters: modelSchema(tool.name, tool.inputSchema),
}));
return [...listed, REMEMBER_SPEC];
return this.offersNotes ? [...listed, REMEMBER_SPEC] : listed;
}
kindOf(name: string): AssistantToolKind | null {
if (name === REMEMBER_TOOL) return 'write';
if (name === REMEMBER_TOOL) return this.offersNotes ? 'write' : null;
return this.tools.get(name)?.kind ?? null;
}
@@ -243,7 +245,8 @@ export class AssistantToolsService implements OnApplicationBootstrap {
pinned: PINNED_ARG in properties,
});
}
return new AssistantToolSession(client, server, tools, tripId, (toolName) => this.audit(userId, toolName));
const offersNotes = include === ASSISTANT_TOOL_ALLOWLIST;
return new AssistantToolSession(client, server, tools, tripId, (name) => this.audit(userId, name), offersNotes);
}
/** The audit row an MCP client's call writes, marked as the assistant's. A failed write never fails the call. */
server/src/nest/assistant/assistant.service.ts
+95 -97
View File
@@ -35,6 +35,7 @@ import { runAgentLoop, type AgentStore, type HeldCall, type StepRecord } from '.
import { AssistantBookingImportService } from './assistant-booking-import.service';
import { mentionItems } from './assistant-cards.helpers';
import { buildTripCard, categoriesFrom, languageName, todayIn, turnContext, type TripCard } from './assistant-context';
import { DISCARDED, NOT_RUN, type Outcome, withCancelled, withOutcomes } from './assistant-decisions.helpers';
import { messageIdOf, outputText, partsOf, toChatHistory, toThreadMessages } from './assistant-history';
import { rememberForTrip } from './assistant-memory';
import { toolOutputForProvider } from './assistant-redact';
@@ -84,22 +85,18 @@ const THREAD_ROWS = 50;
/** Message rows a turn sends the provider. */
const HISTORY_ROWS = 120;
/** What the model and the thread record for a change the user discarded or moved on from. */
const DISCARDED = { applied: false, reason: 'The user discarded this change.' };
const CANCELLED = { applied: false, reason: 'The user moved on without deciding.' };
const NOT_RUN = { applied: false, reason: 'The change could not be run.' };
/** What the read step of an attached file answers when the file could not be read. */
const FILE_UNREADABLE = 'The file could not be read.';
/** The longest file name a message keeps. */
const MAX_FILE_NAME_CHARS = 255;
/** What became of one decided change, as its message records it. */
interface Outcome {
approved: boolean;
result: unknown;
isError: boolean;
/** Where one applied change reports to: the stream, and what its message records. */
interface AppliedSink {
w: UiStreamWriter;
unrun: Set<string>;
outcomes: Map<string, Outcome>;
receipts: AssistantMessagePart[];
}
/**
@@ -624,20 +621,21 @@ export class AssistantService {
private async cancelOpen(threadId: number, now: string): Promise<void> {
const open = await this.toolCalls.listOpen(threadId);
const cancelled = new Set<string>();
for (const call of open) {
if (await this.toolCalls.claimDecision(call.id, threadId, false, dbNow(new Date(0)), now)) cancelled.add(call.id);
}
for (const messageId of new Set(open.map((call) => call.message_id))) {
const row = await this.messages.findInThread(messageId, threadId);
if (!row) continue;
const parts = partsOf(row).map((part) =>
part.type === 'tool-call' && cancelled.has(part.toolCallId)
? { ...part, result: CANCELLED, approval: { id: part.toolCallId, resolution: 'cancelled' as const } }
: part,
);
await this.messages.update(messageId, { parts: encodeParts(parts), status: 'complete' });
}
const claimed = await Promise.all(
open.map((call) => this.toolCalls.claimDecision(call.id, threadId, false, dbNow(new Date(0)), now)),
);
const cancelled = new Set(open.filter((_call, i) => claimed[i]).map((call) => call.id));
await Promise.all(
[...new Set(open.map((call) => call.message_id))].map((id) => this.markCancelled(id, threadId, cancelled)),
);
}
/** Marks the cancelled changes of one message as such, and the message as done. */
private async markCancelled(messageId: number, threadId: number, cancelled: ReadonlySet<string>): Promise<void> {
const row = await this.messages.findInThread(messageId, threadId);
if (!row) return;
const parts = withCancelled(partsOf(row), cancelled);
await this.messages.update(messageId, { parts: encodeParts(parts), status: 'complete' });
}
/**
@@ -660,14 +658,13 @@ export class AssistantService {
const now = new Date();
const notBefore = dbNow(new Date(now.getTime() - APPROVAL_TTL_MS));
const decided = await this.uow.transactional(async () => {
const taken: Array<{ call: AssistantToolCallRow; approved: boolean }> = [];
for (const call of open) {
const approved = wanted.get(call.id) === true;
if (await this.toolCalls.claimDecision(call.id, threadId, approved, notBefore, dbNow(now))) {
taken.push({ call, approved });
}
}
return taken;
const claims = open.map((call) => ({ call, approved: wanted.get(call.id) === true }));
const taken = await Promise.all(
claims.map(({ call, approved }) =>
this.toolCalls.claimDecision(call.id, threadId, approved, notBefore, dbNow(now)),
),
);
return claims.filter((_claim, i) => taken[i]);
});
if (decided.length === 0) {
throw conflict('These changes were decided already or have expired', { code: 'generic' });
@@ -694,49 +691,61 @@ export class AssistantService {
): Promise<void> {
const outcomes = new Map<string, Outcome>();
const receipts: AssistantMessagePart[] = [];
for (const { call, approved } of decided) {
if (!approved) {
w.toolDenied(call.id);
outcomes.set(call.id, { approved: false, result: DISCARDED, isError: false });
continue;
}
const args = decodeJson(ASSISTANT_TOOL_ARGS, call.args, `tool call ${call.id}`);
const ran = await run(call.tool_name, args);
const output = toolOutputForProvider(call.tool_name, ran.value);
const undoable = ran.ok && !!ASSISTANT_UNDO_MAP[writeToolOf(call.tool_name, args)]?.args(args, ran.value);
const undoUntil = undoable ? dbNow(new Date(Date.now() + UNDO_WINDOW_MS)) : null;
await this.inContext(() =>
this.toolCalls.recordOutcome(
call.id,
ran.ok ? 'executed' : 'failed',
encodeJson(ASSISTANT_TOOL_RESULT, ran.value),
undoUntil,
),
);
unrun.delete(call.id);
outcomes.set(call.id, { approved: true, result: output, isError: !ran.ok });
if (!ran.ok) {
w.toolError(call.id, outputText(output));
continue;
}
w.toolOutput(call.id, output);
const preview = decodeJson(ASSISTANT_TOOL_PREVIEW, call.preview, `tool call ${call.id}`);
const applied = appliedChange(
call.tool_name,
call.id,
args,
ran.value,
preview?.dayId ?? null,
utcSuffix(undoUntil),
);
if (applied) {
w.applied(applied);
receipts.push({ type: 'data-trek-applied', data: applied });
}
}
// One after another: a change may build on the one before it.
await decided.reduce<Promise<void>>(async (previous, { call, approved }) => {
await previous;
await this.applyOne(run, call, approved, { w, unrun, outcomes, receipts });
}, Promise.resolve());
await this.recordDecided(threadId, decided[0].call.message_id, outcomes, receipts);
}
/** Runs one decided change, or records it as discarded, and tells the client what became of it. */
private async applyOne(
run: (name: string, args: Record<string, unknown>) => Promise<ToolRun>,
call: AssistantToolCallRow,
approved: boolean,
{ w, unrun, outcomes, receipts }: AppliedSink,
): Promise<void> {
if (!approved) {
w.toolDenied(call.id);
outcomes.set(call.id, { approved: false, result: DISCARDED, isError: false });
return;
}
const args = decodeJson(ASSISTANT_TOOL_ARGS, call.args, `tool call ${call.id}`);
const ran = await run(call.tool_name, args);
const output = toolOutputForProvider(call.tool_name, ran.value);
const undoable = ran.ok && !!ASSISTANT_UNDO_MAP[writeToolOf(call.tool_name, args)]?.args(args, ran.value);
const undoUntil = undoable ? dbNow(new Date(Date.now() + UNDO_WINDOW_MS)) : null;
await this.inContext(() =>
this.toolCalls.recordOutcome(
call.id,
ran.ok ? 'executed' : 'failed',
encodeJson(ASSISTANT_TOOL_RESULT, ran.value),
undoUntil,
),
);
unrun.delete(call.id);
outcomes.set(call.id, { approved: true, result: output, isError: !ran.ok });
if (!ran.ok) {
w.toolError(call.id, outputText(output));
return;
}
w.toolOutput(call.id, output);
const preview = decodeJson(ASSISTANT_TOOL_PREVIEW, call.preview, `tool call ${call.id}`);
const applied = appliedChange(
call.tool_name,
call.id,
args,
ran.value,
preview?.dayId ?? null,
utcSuffix(undoUntil),
);
if (applied) {
w.applied(applied);
receipts.push({ type: 'data-trek-applied', data: applied });
}
}
/**
* The approved changes the turn ended before running (the tool session did
* not open, the stream broke): recorded as failed, so the thread shows them
@@ -751,16 +760,17 @@ export class AssistantService {
): Promise<void> {
const outcomes = new Map<string, Outcome>();
for (const { call, approved } of decided) {
if (!approved) {
outcomes.set(call.id, { approved: false, result: DISCARDED, isError: false });
continue;
}
if (!unrun.has(call.id)) continue;
await this.inContext(() =>
this.toolCalls.recordOutcome(call.id, 'failed', encodeJson(ASSISTANT_TOOL_RESULT, NOT_RUN), null),
);
outcomes.set(call.id, { approved: true, result: NOT_RUN, isError: true });
if (!approved) outcomes.set(call.id, { approved: false, result: DISCARDED, isError: false });
else if (unrun.has(call.id)) outcomes.set(call.id, { approved: true, result: NOT_RUN, isError: true });
}
const failed = decided.filter(({ call, approved }) => approved && unrun.has(call.id));
await Promise.all(
failed.map(({ call }) =>
this.inContext(() =>
this.toolCalls.recordOutcome(call.id, 'failed', encodeJson(ASSISTANT_TOOL_RESULT, NOT_RUN), null),
),
),
);
await this.recordDecided(threadId, decided[0].call.message_id, outcomes, []);
}
@@ -774,16 +784,7 @@ export class AssistantService {
await this.inContext(async () => {
const row = await this.messages.findInThread(messageId, threadId);
if (!row) return;
const parts = partsOf(row).map((part) => {
const outcome = part.type === 'tool-call' ? outcomes.get(part.toolCallId) : undefined;
if (part.type !== 'tool-call' || !outcome) return part;
return {
...part,
result: outcome.result,
...(outcome.isError ? { isError: true } : {}),
approval: { id: part.toolCallId, approved: outcome.approved },
};
});
const parts = withOutcomes(partsOf(row), outcomes);
await this.messages.update(messageId, { parts: encodeParts([...parts, ...receipts]), status: 'complete' });
});
}
@@ -799,12 +800,9 @@ export class AssistantService {
/** The tool call rows behind the held changes of these messages, by id. */
private async callsOf(rows: readonly AssistantMessageRow[]): Promise<Map<string, AssistantToolCallRow>> {
const calls = new Map<string, AssistantToolCallRow>();
for (const row of rows) {
if (!partsOf(row).some((part) => part.type === 'tool-call' && part.approval)) continue;
for (const call of await this.toolCalls.listByMessage(row.id)) calls.set(call.id, call);
}
return calls;
const held = rows.filter((row) => partsOf(row).some((part) => part.type === 'tool-call' && part.approval));
const lists = await Promise.all(held.map((row) => this.toolCalls.listByMessage(row.id)));
return new Map(lists.flat().map((call) => [call.id, call]));
}
/**
server/src/nest/oauth/oauth-cimd.ts
+4 -2
View File
@@ -117,8 +117,10 @@ export class CimdClientResolver {
if (cached && cached.expiresAt > Date.now()) return cached.client;
const pending = this.inflight.get(clientId);
if (pending) return pending;
const lookup = this.lookup(url).finally(() => this.inflight.delete(clientId));
if (pending !== undefined) return pending;
const lookup = this.lookup(url).finally(() => {
this.inflight.delete(clientId);
});
this.inflight.set(clientId, lookup);
return lookup;
}