mirror of
https://github.com/mauriceboe/TREK.git
synced 2026-10-24 07:53:09 +00:00
fix(assistant): a change left waiting past its day no longer holds the next one back
The open changes of a thread are now the ones still inside their decision window. A call older than a day already reads as expired in the thread and can no longer be decided, but it stayed approval_requested until the nightly sweep, so "+ Tour", "+ Day" and "Plan this" answered decide_first with nothing left to decide, and a decision had to name the expired call too. listOpen takes the window's start, and every caller passes the same cutoff.
This commit is contained in:
@@ -145,13 +145,18 @@ export class AssistantToolCallsRepository extends TrekRepository<AssistantToolCa
|
||||
.execute();
|
||||
}
|
||||
|
||||
/** The calls of a thread still waiting for a decision. */
|
||||
async listOpen(threadId: number): Promise<AssistantToolCallRow[]> {
|
||||
/**
|
||||
* The calls of a thread still waiting for a decision, proposed at or after
|
||||
* `notBefore`: an older one has expired, as the thread shows it and as
|
||||
* `claimDecision` refuses it, even before the nightly sweep marks it so.
|
||||
*/
|
||||
async listOpen(threadId: number, notBefore: string): Promise<AssistantToolCallRow[]> {
|
||||
return this.db()
|
||||
.selectFrom('assistant_tool_calls')
|
||||
.select(COLUMNS)
|
||||
.where('thread_id', '=', threadId)
|
||||
.where('state', '=', 'approval_requested')
|
||||
.where('created_at', '>=', notBefore)
|
||||
.orderBy('created_at')
|
||||
.orderBy('id')
|
||||
.execute();
|
||||
|
||||
@@ -30,7 +30,7 @@ import {
|
||||
} from './assistant-context';
|
||||
import { partsOf, toThreadMessages } from './assistant-history';
|
||||
import { AssistantStreamRegistry } from './assistant-stream.registry';
|
||||
import { ADD_FOUND_PLACE_TOOL, TURNS_PER_MINUTE } from './assistant-tools.constants';
|
||||
import { ADD_FOUND_PLACE_TOOL, decisionCutoff, TURNS_PER_MINUTE } from './assistant-tools.constants';
|
||||
import { AssistantToolsService } from './assistant-tools.service';
|
||||
import { AssistantCardContext } from './cards/card-context';
|
||||
import { recordOf } from './cards/card-values';
|
||||
@@ -248,7 +248,7 @@ export class AssistantFoundPlacesService {
|
||||
tourId: number | null,
|
||||
argsOf: ArgsOf,
|
||||
): Promise<void> {
|
||||
if ((await this.toolCalls.listOpen(threadId)).length > 0) {
|
||||
if ((await this.toolCalls.listOpen(threadId, decisionCutoff())).length > 0) {
|
||||
throw aiRefusal('decide_first', 'Decide on the changes that are waiting first');
|
||||
}
|
||||
const place = foundPlace(await this.messages.listLatest(threadId, SEARCHED_ROWS), req.ref);
|
||||
|
||||
@@ -25,7 +25,7 @@ import { partsOf, toThreadMessages } from './assistant-history';
|
||||
import { isSportPlan, SPORT_CREATES, sportTakeArgsOf, tourTypeWord } from './assistant-proposal-sports';
|
||||
import { isWayProposal, WAY_CREATES, wayTakeArgsOf } from './assistant-proposal-ways';
|
||||
import { AssistantStreamRegistry } from './assistant-stream.registry';
|
||||
import { TAKE_PROPOSAL_TOOL, TURNS_PER_MINUTE } from './assistant-tools.constants';
|
||||
import { decisionCutoff, TAKE_PROPOSAL_TOOL, TURNS_PER_MINUTE } from './assistant-tools.constants';
|
||||
import { AssistantToolsService, type ToolRun } from './assistant-tools.service';
|
||||
import { AssistantCardContext } from './cards/card-context';
|
||||
import { recordOf, type Row } from './cards/card-values';
|
||||
@@ -276,7 +276,7 @@ export class AssistantProposalsService {
|
||||
}
|
||||
|
||||
private async hold(userId: number, tripId: number, threadId: number, req: AssistantTakeProposalRequest) {
|
||||
if ((await this.toolCalls.listOpen(threadId)).length > 0) {
|
||||
if ((await this.toolCalls.listOpen(threadId, decisionCutoff())).length > 0) {
|
||||
throw aiRefusal('decide_first', 'Decide on the changes that are waiting first');
|
||||
}
|
||||
const stored = storedProposal(await this.messages.listLatest(threadId, SEARCHED_ROWS), req.toolCallId);
|
||||
|
||||
@@ -7,7 +7,7 @@ import { AiConnectionsService } from '../ai-connections/ai-connections.service';
|
||||
import { AiUsageService } from '../ai-connections/ai-usage.service';
|
||||
import { logError, logInfo } from '../audit/audit-log.logger';
|
||||
import { CronRegistrarService } from '../scheduling/cron-registrar.service';
|
||||
import { APPROVAL_TTL_MS } from './assistant-tools.constants';
|
||||
import { decisionCutoff } from './assistant-tools.constants';
|
||||
import { InjectRepository } from '@mikro-orm/nestjs';
|
||||
import { Injectable, type OnApplicationBootstrap } from '@nestjs/common';
|
||||
|
||||
@@ -51,7 +51,7 @@ export class AssistantRetentionJob implements OnApplicationBootstrap {
|
||||
: 0;
|
||||
const orphaned = await this.threads.deleteWithoutAccess();
|
||||
if (idle + orphaned > 0) logInfo(`Assistant retention: removed ${idle + orphaned} conversation(s)`);
|
||||
const expired = await this.toolCalls.expireBefore(dbNow(new Date(now.getTime() - APPROVAL_TTL_MS)));
|
||||
const expired = await this.toolCalls.expireBefore(decisionCutoff(now));
|
||||
if (expired.length > 0) logInfo(`Assistant retention: ${expired.length} proposed change(s) expired`);
|
||||
await this.usage.purgeOlderThan(ASSISTANT_USAGE_RETENTION_DAYS);
|
||||
} catch (err: unknown) {
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { dbNow, type DbTimestamp } from '../../db/types';
|
||||
import type { ToolUndo } from './cards/card-types';
|
||||
import { argId, idIn } from './cards/card-values';
|
||||
import type { AssistantApplied, AssistantChangeVerbKey } from '@trek/shared';
|
||||
@@ -27,6 +28,11 @@ export const UNDO_WINDOW_MS = 15 * 60_000;
|
||||
/** How long a proposed change waits for a decision before it expires. */
|
||||
export const APPROVAL_TTL_MS = 24 * 60 * 60_000;
|
||||
|
||||
/** The oldest a change still waiting at `now` may have been proposed and still be decided, as the database writes it. */
|
||||
export function decisionCutoff(now: Date = new Date()): DbTimestamp {
|
||||
return dbNow(new Date(now.getTime() - APPROVAL_TTL_MS));
|
||||
}
|
||||
|
||||
/** Turns (chat, decisions, undo) a user may start per minute. */
|
||||
export const TURNS_PER_MINUTE = 30;
|
||||
|
||||
|
||||
@@ -45,8 +45,8 @@ import { AssistantScopes, findScopedCall, scopedThreads, tripScope } from './ass
|
||||
import { AssistantStreamRegistry } from './assistant-stream.registry';
|
||||
import { titleFrom } from './assistant-title';
|
||||
import {
|
||||
APPROVAL_TTL_MS,
|
||||
ASSISTANT_SERVER_WRITE_TOOLS,
|
||||
decisionCutoff,
|
||||
foundPlaceArgs,
|
||||
IMPORT_BOOKING_TOOL,
|
||||
REMEMBER_TOOL,
|
||||
@@ -316,7 +316,7 @@ export class AssistantService {
|
||||
): Promise<void> {
|
||||
if (!(await scopedThreads(this.threads, scope).find(threadId, userId))) throw threadNotFound();
|
||||
await this.admit(userId);
|
||||
const open = await this.toolCalls.listOpen(threadId);
|
||||
const open = await this.toolCalls.listOpen(threadId, decisionCutoff());
|
||||
if (open.length > 0 && open.every((call) => ASSISTANT_SERVER_WRITE_TOOLS[call.tool_name]?.quiet === true)) {
|
||||
await this.decideQuietly(userId, scope, threadId, req, channel);
|
||||
return;
|
||||
@@ -624,7 +624,7 @@ export class AssistantService {
|
||||
}
|
||||
|
||||
private async cancelOpen(threadId: number, now: string): Promise<void> {
|
||||
const open = await this.toolCalls.listOpen(threadId);
|
||||
const open = await this.toolCalls.listOpen(threadId, decisionCutoff());
|
||||
const claimed = await Promise.all(
|
||||
open.map((call) => this.toolCalls.claimDecision(call.id, threadId, false, dbNow(new Date(0)), now)),
|
||||
);
|
||||
@@ -644,14 +644,16 @@ export class AssistantService {
|
||||
|
||||
/**
|
||||
* Records the decisions, each only while its change still waits and is less
|
||||
* than a day old. The request must name every waiting change and nothing
|
||||
* else (422); when none could be taken any more, 409.
|
||||
* than a day old. The request must name every change waiting within its day
|
||||
* and nothing else (422); when none could be taken any more, 409.
|
||||
*/
|
||||
private async claimDecisions(
|
||||
threadId: number,
|
||||
decisions: AssistantDecision[],
|
||||
): Promise<Array<{ call: AssistantToolCallRow; approved: boolean }>> {
|
||||
const open = await this.toolCalls.listOpen(threadId);
|
||||
const now = new Date();
|
||||
const notBefore = decisionCutoff(now);
|
||||
const open = await this.toolCalls.listOpen(threadId, notBefore);
|
||||
if (open.length === 0) throw conflict('No change is waiting for a decision', { code: 'generic' });
|
||||
const wanted = new Map(decisions.map((decision) => [decision.toolCallId, decision.approved]));
|
||||
const complete =
|
||||
@@ -659,8 +661,6 @@ export class AssistantService {
|
||||
if (!complete) {
|
||||
throw new DomainError(422, 'Decide on every change that is waiting', { details: { code: 'generic' } });
|
||||
}
|
||||
const now = new Date();
|
||||
const notBefore = dbNow(new Date(now.getTime() - APPROVAL_TTL_MS));
|
||||
// In the order the decisions name them, the order the chat shows the changes in: the changes of one step
|
||||
// share their time, and their ids sort at random, so only this order lets "undo all" walk them back.
|
||||
const position = new Map(decisions.map((decision, i) => [decision.toolCallId, i]));
|
||||
|
||||
@@ -27,6 +27,7 @@ import { DayAssignments } from '../../src/db/entities/DayAssignments.entity';
|
||||
import { Places } from '../../src/db/entities/Places.entity';
|
||||
import { Reservations } from '../../src/db/entities/Reservations.entity';
|
||||
import { Trips } from '../../src/db/entities/Trips.entity';
|
||||
import { dbNow } from '../../src/db/types';
|
||||
import { TURNS_PER_MINUTE } from '../../src/nest/assistant/assistant-tools.constants';
|
||||
import { AssistantToolsService } from '../../src/nest/assistant/assistant-tools.service';
|
||||
import { AssistantCardContext } from '../../src/nest/assistant/cards/card-context';
|
||||
@@ -896,6 +897,43 @@ describe('Trip assistant e2e', () => {
|
||||
expect(provider.requests).toHaveLength(2);
|
||||
});
|
||||
|
||||
it('ASTE2E-025: a change left waiting past its day holds nothing back, and the next one is decided alone', async () => {
|
||||
const t = await travelerWithTrip();
|
||||
const [day] = await readTripDays(orm, t.tripId);
|
||||
const { threadId } = await searchTurn(t);
|
||||
const path = `${t.base}/threads/${threadId}/found-places`;
|
||||
// The newest change the answer holds: a found place joins the answer of its search.
|
||||
const changeOf = (res: { body: { message: { parts: AssistantMessagePart[] } } }) =>
|
||||
res.body.message.parts
|
||||
.flatMap((part) => (part.type === 'data-trek-change' ? [part.data.toolCallId] : []))
|
||||
.at(-1) ?? '';
|
||||
const stale = changeOf(await addFound(path, t.userId, { ref: 'call_find:0', dayId: day.id }));
|
||||
// A day and an hour old: the thread shows it expired, and the nightly sweep has not run yet.
|
||||
await updateRows(
|
||||
orm,
|
||||
AssistantToolCalls,
|
||||
{ id: stale },
|
||||
{ created_at: dbNow(new Date(Date.now() - 25 * 3_600_000)) },
|
||||
);
|
||||
const shown = await get(`${t.base}/threads/${threadId}`, t.userId);
|
||||
const parts = (shown.body.messages as Array<{ parts: AssistantMessagePart[] }>).flatMap((m) => m.parts);
|
||||
expect(parts.find((part) => part.type === 'tool-call' && part.toolCallId === stale)).toMatchObject({
|
||||
approval: { resolution: 'expired' },
|
||||
});
|
||||
|
||||
const next = await addFound(path, t.userId, { ref: 'call_find:1', dayId: day.id });
|
||||
expect(next.status).toBe(200);
|
||||
const fresh = changeOf(next);
|
||||
expect(fresh).not.toBe(stale);
|
||||
const decided = await post(`${t.base}/threads/${threadId}/decisions`, t.userId, {
|
||||
...CTX,
|
||||
decisions: [{ toolCallId: fresh, approved: true }],
|
||||
});
|
||||
expect(decided.status).toBe(200);
|
||||
expect(ofType(decided.chunks, 'data-trek-applied')[0].data).toMatchObject({ toolCallId: fresh });
|
||||
expect((await findRow(orm, AssistantToolCalls, { id: stale }))?.state).toBe('approval_requested');
|
||||
});
|
||||
|
||||
it('ASTE2E-024: "+ Day" asks for the right to edit places: 403 and nothing held without it, 404 outside the trip', async () => {
|
||||
const t = await travelerWithTrip();
|
||||
const { user: member } = await makeUser(orm);
|
||||
|
||||
@@ -324,7 +324,9 @@ describe('proposed changes', () => {
|
||||
const { threadId, messageId } = await threadWithMessage();
|
||||
await calls.insertProposals([proposal('c1', threadId, messageId), proposal('c2', threadId, messageId)]);
|
||||
await calls.insertProposals([]);
|
||||
expect((await calls.listOpen(threadId)).map((row) => row.id)).toEqual(['c1', 'c2']);
|
||||
expect((await calls.listOpen(threadId, AT)).map((row) => row.id)).toEqual(['c1', 'c2']);
|
||||
// Proposed before `notBefore`, a change waits no more, though no sweep has marked it expired yet.
|
||||
expect(await calls.listOpen(threadId, LATER)).toEqual([]);
|
||||
|
||||
expect(await calls.claimDecision('c1', threadId + 1, true, AT, LATER)).toBe(false);
|
||||
expect(await calls.claimDecision('c1', threadId, true, '2026-10-21 00:00:00', LATER)).toBe(false);
|
||||
|
||||
@@ -3,7 +3,7 @@
|
||||
* call repositories on a migrated test db: the place comes from the search
|
||||
* results stored in the thread, the day must be one of the trip's, and the
|
||||
* suggestion is held as a change card in one transaction, refused while a turn
|
||||
* runs, while another change waits, or past the per-minute limit.
|
||||
* runs, while another change waits within its day, or past the per-minute limit.
|
||||
*
|
||||
* Replaced: the trip tools (what the trip reads as), the rate limit, the
|
||||
* stream registry and the airports.
|
||||
@@ -11,6 +11,7 @@
|
||||
import { AssistantMessages } from '../../../../src/db/entities/AssistantMessages.entity';
|
||||
import { AssistantThreads } from '../../../../src/db/entities/AssistantThreads.entity';
|
||||
import { AssistantToolCalls } from '../../../../src/db/entities/AssistantToolCalls.entity';
|
||||
import { dbNow } from '../../../../src/db/types';
|
||||
import { AirportsService } from '../../../../src/nest/airports/airports.service';
|
||||
import { AssistantFoundPlacesService } from '../../../../src/nest/assistant/assistant-found-places.service';
|
||||
import { AssistantStreamRegistry } from '../../../../src/nest/assistant/assistant-stream.registry';
|
||||
@@ -19,7 +20,7 @@ import { AssistantCardContext } from '../../../../src/nest/assistant/cards/card-
|
||||
import { RateLimitService } from '../../../../src/nest/common/rate-limit.service';
|
||||
import { UnitOfWork } from '../../../../src/nest/database/unit-of-work';
|
||||
import { createSnapshotTestDb } from '../../../helpers/db-mock';
|
||||
import { createRow, findRows } from '../../../helpers/factories/rows';
|
||||
import { createRow, findRows, updateRows } from '../../../helpers/factories/rows';
|
||||
import { makeTrip } from '../../../helpers/factories/trips';
|
||||
import { makeUser } from '../../../helpers/factories/users';
|
||||
import { resetTestDb } from '../../../helpers/test-db';
|
||||
@@ -182,6 +183,19 @@ describe('AssistantFoundPlacesService', () => {
|
||||
expect(await calls(c.threadId)).toHaveLength(1);
|
||||
});
|
||||
|
||||
it('ASTFND-008: a change left waiting past its day holds nothing back, before the nightly sweep marks it', async () => {
|
||||
const c = await conversation();
|
||||
await add(c, 's1:0');
|
||||
const [stale] = await calls(c.threadId);
|
||||
const dayAndHourAgo = dbNow(new Date(Date.now() - 25 * 60 * 60_000));
|
||||
await updateRows(t.get(MikroORM), AssistantToolCalls, { id: stale?.id ?? '' }, { created_at: dayAndHourAgo });
|
||||
|
||||
await add(c, 's1:1');
|
||||
const held = await calls(c.threadId);
|
||||
expect(held).toHaveLength(2);
|
||||
expect(held.map((call) => call.state)).toEqual(['approval_requested', 'approval_requested']);
|
||||
});
|
||||
|
||||
it('ASTFND-007: a ref outside the search results or a day of another trip proposes nothing', async () => {
|
||||
const c = await conversation();
|
||||
await expect(add(c, 's9:0')).rejects.toMatchObject({ status: 400, code: 'generic' });
|
||||
|
||||
@@ -3,7 +3,7 @@
|
||||
* and tool call repositories on a migrated test db: the proposal comes from
|
||||
* the call and card stored in the thread, never from the request, and the
|
||||
* change that creates it is held as a card in one transaction, refused while
|
||||
* a turn runs, while another change waits, for a proposal the thread does not
|
||||
* a turn runs, while another change waits within its day, for a proposal the thread does not
|
||||
* hold and for one nothing can be created from. A walk is asked again, since
|
||||
* the stored answer is the model's shortened copy.
|
||||
*
|
||||
@@ -13,6 +13,7 @@
|
||||
import { AssistantMessages } from '../../../../src/db/entities/AssistantMessages.entity';
|
||||
import { AssistantThreads } from '../../../../src/db/entities/AssistantThreads.entity';
|
||||
import { AssistantToolCalls } from '../../../../src/db/entities/AssistantToolCalls.entity';
|
||||
import { dbNow } from '../../../../src/db/types';
|
||||
import { AirportsService } from '../../../../src/nest/airports/airports.service';
|
||||
import {
|
||||
AssistantProposalsService,
|
||||
@@ -27,7 +28,7 @@ import { RateLimitService } from '../../../../src/nest/common/rate-limit.service
|
||||
import { UnitOfWork } from '../../../../src/nest/database/unit-of-work';
|
||||
import { tourRouteCard } from '../../../helpers/assistant-tour-cards';
|
||||
import { createSnapshotTestDb } from '../../../helpers/db-mock';
|
||||
import { createRow, findRows } from '../../../helpers/factories/rows';
|
||||
import { createRow, findRows, updateRows } from '../../../helpers/factories/rows';
|
||||
import { makeTrip } from '../../../helpers/factories/trips';
|
||||
import { makeUser } from '../../../helpers/factories/users';
|
||||
import { resetTestDb } from '../../../helpers/test-db';
|
||||
@@ -226,6 +227,19 @@ describe('AssistantProposalsService', () => {
|
||||
expect(await calls(c.threadId)).toHaveLength(1);
|
||||
});
|
||||
|
||||
it('ASTPROP-011: a change left waiting past its day holds nothing back, before the nightly sweep marks it', async () => {
|
||||
const c = await conversation();
|
||||
await take(c, 'h1');
|
||||
const [stale] = await calls(c.threadId);
|
||||
const dayAndHourAgo = dbNow(new Date(Date.now() - 25 * 60 * 60_000));
|
||||
await updateRows(t.get(MikroORM), AssistantToolCalls, { id: stale?.id ?? '' }, { created_at: dayAndHourAgo });
|
||||
|
||||
await take(c, 'w1');
|
||||
const held = await calls(c.threadId);
|
||||
expect(held).toHaveLength(2);
|
||||
expect(held.map((call) => call.state)).toEqual(['approval_requested', 'approval_requested']);
|
||||
});
|
||||
|
||||
it('ASTPROP-008: a walk stored cut down and masked is created from the tool asked again, named and on no day', async () => {
|
||||
const c = await conversation();
|
||||
await take(c, 'w1');
|
||||
|
||||
Reference in New Issue
Block a user