fix(quote-assistant): retry unanswered lines, bound and serialise ticks
- decideWithAi throws on stop_reason max_tokens instead of returning a partial answer; fillQuote marks a requested line the AI did not answer as error (retried after 1 h), never a permanent no_match. - QUOTE_MAX_LINES 30 -> 15; the tick also caps lines per quote so the 60-line budget reaches other quotes. - runAssistantTick stops starting quotes after a 90 s time budget and returns reason running while a newer start than the last summary is under 10 minutes old (lease). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
parent
51bd4ec32c
commit
a9160de1d6
6 changed files with 205 additions and 21 deletions
|
|
@ -77,14 +77,23 @@ Click-by-click:
|
|||
4. Save. Leave the workflow **inactive** until steps 1 and 2 above are done.
|
||||
5. After deploy, click Execute workflow once. Expected response while the
|
||||
assistant is off: `{"ran":false,"reason":"disabled"}`. When on and not
|
||||
yet due: `{"ran":false,"reason":"not_due"}`. When it ran:
|
||||
yet due: `{"ran":false,"reason":"not_due"}`. While an earlier run is
|
||||
still going: `{"ran":false,"reason":"running"}`. When it ran:
|
||||
`{"ran":true,"summary":{...}}`.
|
||||
6. Toggle the workflow Active.
|
||||
|
||||
A 401 means the credential is wrong (it must send the app's
|
||||
`AGENT_API_KEY` as a Bearer token). A tick that takes long is normal when
|
||||
there is work: each run is bounded (at most 20 quotes and 60 lines, newest
|
||||
first, 30 lines per quote), and leftovers carry to the next run.
|
||||
there is work. Each run is bounded: at most 20 quotes and 60 lines, newest
|
||||
first, 15 lines per quote (one AI call per quote). It also stops starting
|
||||
new quotes once 90 seconds have passed, so a run normally answers inside
|
||||
the 120 s timeout; leftovers carry to the next run.
|
||||
|
||||
Runs never overlap. A run stamps its start time before it does any work and
|
||||
writes its summary when it ends. While a start is newer than the last
|
||||
summary and less than 10 minutes old, other ticks answer
|
||||
`{"ran":false,"reason":"running"}`. If a run dies without writing its
|
||||
summary, the next tick after those 10 minutes takes over.
|
||||
|
||||
## Reading `ai_fill`
|
||||
|
||||
|
|
|
|||
|
|
@ -12,7 +12,7 @@ import { runAssistantTick } from "@/lib/server/quote-assistant-service";
|
|||
* Auth: `assertAgentAuth` (Bearer `AGENT_API_KEY`).
|
||||
*
|
||||
* Response: the `runAssistantTick` result, e.g.
|
||||
* { ran: false, reason: "disabled" | "not_due" } or { ran: true, summary }
|
||||
* { ran: false, reason: "disabled" | "not_due" | "running" } or { ran: true, summary }
|
||||
*/
|
||||
export const runtime = "nodejs";
|
||||
export const dynamic = "force-dynamic";
|
||||
|
|
|
|||
|
|
@ -160,6 +160,29 @@ describe("decideWithAi", () => {
|
|||
expect(res.decisions.get("qi2")?.stock_item_id).toBeNull();
|
||||
});
|
||||
|
||||
it("throws when the response was cut off by max_tokens, even with a partial tool call", async () => {
|
||||
const { client, create } = fakeClient([
|
||||
toolUse({
|
||||
matches: [
|
||||
{ quote_item_id: "qi1", stock_item_id: "stk1", confidence: "high", reason: "", doubts: [] },
|
||||
],
|
||||
}),
|
||||
]);
|
||||
create.mockResolvedValueOnce({
|
||||
content: [
|
||||
toolUse({
|
||||
matches: [
|
||||
{ quote_item_id: "qi1", stock_item_id: "stk1", confidence: "high", reason: "", doubts: [] },
|
||||
],
|
||||
}),
|
||||
],
|
||||
stop_reason: "max_tokens",
|
||||
});
|
||||
await expect(decideWithAi(input, { client })).rejects.toThrow(
|
||||
/^AI response was cut off \(max_tokens\)/,
|
||||
);
|
||||
});
|
||||
|
||||
it("throws when the response has no submit_line_matches tool call", async () => {
|
||||
const { client } = fakeClient([{ type: "text", text: "sorry" }]);
|
||||
await expect(decideWithAi(input, { client })).rejects.toThrow(
|
||||
|
|
|
|||
|
|
@ -134,6 +134,15 @@ export async function decideWithAi(
|
|||
messages: [{ role: "user", content: buildDecideUserText(input) }],
|
||||
});
|
||||
|
||||
// A truncated tool call silently drops the lines it never reached; fail
|
||||
// the whole call so those lines are marked `error` and retried, rather
|
||||
// than recorded as permanent no-matches.
|
||||
if (response.stop_reason === "max_tokens") {
|
||||
throw new Error(
|
||||
`AI response was cut off (max_tokens) at ${aiConfig.maxTokens} tokens before all ${input.lines.length} lines were answered`,
|
||||
);
|
||||
}
|
||||
|
||||
const toolBlock = response.content.find(
|
||||
(b) => b.type === "tool_use" && b.name === TOOL_NAME,
|
||||
);
|
||||
|
|
|
|||
|
|
@ -77,6 +77,7 @@ const settings = {
|
|||
quote_assistant_interval_minutes: 10,
|
||||
quote_assistant_min_confidence: "medium" as "high" | "medium",
|
||||
quote_assistant_last_run_at: "",
|
||||
quote_assistant_last_run_summary: null as null | { started_at: string; finished_at: string },
|
||||
};
|
||||
const recordRun = vi.fn(async (fields: { last_run_at?: string; summary?: unknown }) => {
|
||||
events.push(`record:${Object.keys(fields).join(",")}`);
|
||||
|
|
@ -99,6 +100,7 @@ import {
|
|||
runAssistantTick,
|
||||
TICK_MAX_LINES,
|
||||
TICK_MAX_QUOTES,
|
||||
TICK_TIME_BUDGET_MS,
|
||||
type AssistantDeps,
|
||||
} from "./quote-assistant-service";
|
||||
|
||||
|
|
@ -199,6 +201,7 @@ beforeEach(() => {
|
|||
quote_assistant_interval_minutes: 10,
|
||||
quote_assistant_min_confidence: "medium",
|
||||
quote_assistant_last_run_at: "",
|
||||
quote_assistant_last_run_summary: null,
|
||||
});
|
||||
});
|
||||
|
||||
|
|
@ -321,8 +324,8 @@ describe("fillQuote", () => {
|
|||
await fillQuote("q1", { deps });
|
||||
|
||||
const sent = decide.mock.calls[0][0].lines.map((l) => l.quote_item_id);
|
||||
expect(QUOTE_MAX_LINES).toBe(30);
|
||||
expect(sent).toHaveLength(30);
|
||||
expect(QUOTE_MAX_LINES).toBe(15);
|
||||
expect(sent).toHaveLength(15);
|
||||
expect(sent).toContain("l39");
|
||||
expect(sent).not.toContain("l0");
|
||||
});
|
||||
|
|
@ -392,15 +395,45 @@ describe("fillQuote", () => {
|
|||
expect(store.items.get("a")!.ai_fill).toBeNull();
|
||||
});
|
||||
|
||||
it("writes no_match for a line the AI returned no decision for", async () => {
|
||||
it("marks a line the AI returned no decision for as error (retried later), not no_match", async () => {
|
||||
addQuote("q1");
|
||||
const a = addItem("q1", { id: "a" });
|
||||
const { deps } = makeDeps({ decisions: () => [] });
|
||||
const b = addItem("q1", { id: "b" });
|
||||
// The AI answered for b only (e.g. it skipped a line).
|
||||
const { deps } = makeDeps({ decisions: () => [decisionFor(b)] });
|
||||
const r = await fillQuote("q1", { deps });
|
||||
expect(r).toMatchObject({ filled: 0, unmatched: 1 });
|
||||
expect(r).toMatchObject({ filled: 1, unmatched: 0 });
|
||||
expect(r.error).toContain("no decision for 1 line");
|
||||
const row = store.items.get(a)!;
|
||||
expect((row.ai_fill as { status: string }).status).toBe("error");
|
||||
expect(row.cost_price).toBe(0);
|
||||
expect(row.pricing_notes).toBe("");
|
||||
});
|
||||
|
||||
it("keeps no_match for a returned decision whose stock_item_id is null", async () => {
|
||||
addQuote("q1");
|
||||
const a = addItem("q1", { id: "a" });
|
||||
const { deps } = makeDeps({
|
||||
decisions: () => [decisionFor(a, { stock_item_id: null, confidence: "low" })],
|
||||
});
|
||||
const r = await fillQuote("q1", { deps });
|
||||
expect(r).toMatchObject({ filled: 0, unmatched: 1, error: "" });
|
||||
expect((store.items.get(a)!.ai_fill as { status: string }).status).toBe("no_match");
|
||||
});
|
||||
|
||||
it("marks every requested line error when the AI response was cut off (max_tokens)", async () => {
|
||||
addQuote("q1");
|
||||
const a = addItem("q1", { id: "a" });
|
||||
const { deps } = makeDeps({
|
||||
decide: async () => {
|
||||
throw new Error("AI response was cut off (max_tokens) before all lines were answered");
|
||||
},
|
||||
});
|
||||
const r = await fillQuote("q1", { deps });
|
||||
expect(r.error).toContain("max_tokens");
|
||||
expect((store.items.get(a)!.ai_fill as { status: string }).status).toBe("error");
|
||||
});
|
||||
|
||||
it("does not write a fill below the configured minimum confidence", async () => {
|
||||
settings.quote_assistant_min_confidence = "high";
|
||||
addQuote("q1");
|
||||
|
|
@ -557,12 +590,10 @@ describe("runAssistantTick", () => {
|
|||
});
|
||||
|
||||
it("processes at most TICK_MAX_LINES lines, newest first", async () => {
|
||||
addQuote("q1");
|
||||
addQuote("q2");
|
||||
addQuote("q3");
|
||||
for (let i = 0; i < 30; i++) addItem("q1", { id: `a${i}` });
|
||||
for (let i = 0; i < 30; i++) addItem("q2", { id: `b${i}` });
|
||||
for (let i = 0; i < 30; i++) addItem("q3", { id: `c${i}` });
|
||||
for (const q of ["a", "b", "c", "d", "e"]) {
|
||||
addQuote(q);
|
||||
for (let i = 0; i < 15; i++) addItem(q, { id: `${q}${i}` });
|
||||
}
|
||||
const { deps, decide } = makeDeps();
|
||||
|
||||
const out = await runAssistantTick(deps);
|
||||
|
|
@ -570,11 +601,88 @@ describe("runAssistantTick", () => {
|
|||
expect(TICK_MAX_LINES).toBe(60);
|
||||
const sent = decide.mock.calls.flatMap((c) => c[0].lines.map((l) => l.quote_item_id));
|
||||
expect(sent).toHaveLength(60);
|
||||
// q3 and q2 are the newest; q1 (oldest) waits for the next tick.
|
||||
expect(sent.every((id) => id.startsWith("c") || id.startsWith("b"))).toBe(true);
|
||||
// e..b are the newest; a (oldest) waits for the next tick.
|
||||
expect(sent.some((id) => id.startsWith("a"))).toBe(false);
|
||||
expect(out.summary!.lines_filled).toBe(60);
|
||||
});
|
||||
|
||||
it("takes at most QUOTE_MAX_LINES lines per quote so the rest of the budget goes to other quotes", async () => {
|
||||
addQuote("q1");
|
||||
addQuote("q2");
|
||||
for (let i = 0; i < 5; i++) addItem("q1", { id: `a${i}` });
|
||||
for (let i = 0; i < 70; i++) addItem("q2", { id: `b${i}` });
|
||||
const { deps, decide } = makeDeps();
|
||||
|
||||
await runAssistantTick(deps);
|
||||
|
||||
const sent = decide.mock.calls.flatMap((c) => c[0].lines.map((l) => l.quote_item_id));
|
||||
expect(sent.filter((id) => id.startsWith("b"))).toHaveLength(QUOTE_MAX_LINES);
|
||||
expect(sent.filter((id) => id.startsWith("a"))).toHaveLength(5);
|
||||
});
|
||||
|
||||
it("stops starting new quotes once the tick time budget is spent; leftovers wait", async () => {
|
||||
addQuote("q1");
|
||||
addQuote("q2");
|
||||
addQuote("q3");
|
||||
addItem("q1");
|
||||
addItem("q2");
|
||||
addItem("q3");
|
||||
let clock = NOW.getTime();
|
||||
const { deps, decide } = makeDeps();
|
||||
deps.now = () => new Date(clock);
|
||||
decide.mockImplementation(async (input) => {
|
||||
clock += 50_000; // each AI call takes 50 s
|
||||
return {
|
||||
decisions: new Map(input.lines.map((l) => [l.quote_item_id, decisionFor(l.quote_item_id)])),
|
||||
model: "m",
|
||||
};
|
||||
});
|
||||
|
||||
const out = await runAssistantTick(deps);
|
||||
|
||||
expect(TICK_TIME_BUDGET_MS).toBe(90_000);
|
||||
// Quote 1 starts at 0 s, quote 2 at 50 s, quote 3 would start at 100 s.
|
||||
expect(decide).toHaveBeenCalledTimes(2);
|
||||
expect(out.summary!.quotes_seen).toBe(2);
|
||||
expect(out.summary!.lines_filled).toBe(2);
|
||||
});
|
||||
|
||||
it("returns running when the previous run started under 10 minutes ago and has not finished", async () => {
|
||||
settings.quote_assistant_interval_minutes = 5;
|
||||
settings.quote_assistant_last_run_at = new Date(NOW.getTime() - 6 * 60_000).toISOString();
|
||||
settings.quote_assistant_last_run_summary = {
|
||||
started_at: new Date(NOW.getTime() - 20 * 60_000).toISOString(),
|
||||
finished_at: new Date(NOW.getTime() - 19 * 60_000).toISOString(),
|
||||
};
|
||||
const { deps, decide } = makeDeps();
|
||||
expect(await runAssistantTick(deps)).toEqual({ ran: false, reason: "running" });
|
||||
expect(recordRun).not.toHaveBeenCalled();
|
||||
expect(decide).not.toHaveBeenCalled();
|
||||
|
||||
// No summary at all (first run still going) counts as running too.
|
||||
settings.quote_assistant_last_run_summary = null;
|
||||
expect(await runAssistantTick(deps)).toEqual({ ran: false, reason: "running" });
|
||||
});
|
||||
|
||||
it("runs when the previous run finished", async () => {
|
||||
settings.quote_assistant_interval_minutes = 5;
|
||||
const last = new Date(NOW.getTime() - 6 * 60_000).toISOString();
|
||||
settings.quote_assistant_last_run_at = last;
|
||||
settings.quote_assistant_last_run_summary = {
|
||||
started_at: last,
|
||||
finished_at: new Date(NOW.getTime() - 5.5 * 60_000).toISOString(),
|
||||
};
|
||||
const { deps } = makeDeps();
|
||||
expect((await runAssistantTick(deps)).ran).toBe(true);
|
||||
});
|
||||
|
||||
it("takes over an unfinished run whose lease (10 minutes) has expired", async () => {
|
||||
settings.quote_assistant_last_run_at = new Date(NOW.getTime() - 11 * 60_000).toISOString();
|
||||
settings.quote_assistant_last_run_summary = null;
|
||||
const { deps } = makeDeps();
|
||||
expect((await runAssistantTick(deps)).ran).toBe(true);
|
||||
});
|
||||
|
||||
it("does not resend an unmatched line whose description is unchanged", async () => {
|
||||
addQuote("q1");
|
||||
addItem("q1", {
|
||||
|
|
|
|||
|
|
@ -57,7 +57,12 @@ export interface FillQuoteResult {
|
|||
|
||||
export const TICK_MAX_QUOTES = 20;
|
||||
export const TICK_MAX_LINES = 60;
|
||||
export const QUOTE_MAX_LINES = 30;
|
||||
/** Lines per AI call: keeps one call well inside the model's max_tokens. */
|
||||
export const QUOTE_MAX_LINES = 15;
|
||||
/** Stop starting new quotes after this long; leftovers go to the next run. */
|
||||
export const TICK_TIME_BUDGET_MS = 90_000;
|
||||
/** A started run with no newer summary is treated as in progress this long. */
|
||||
export const TICK_LEASE_MS = 10 * 60_000;
|
||||
|
||||
const ACTOR_NAME = "AI quote assistant";
|
||||
const COUNT_CHUNK = 50;
|
||||
|
|
@ -161,8 +166,13 @@ export async function fillQuote(
|
|||
}
|
||||
|
||||
// 3. Re-read each line and write its patch.
|
||||
let unanswered = 0;
|
||||
for (const line of lines) {
|
||||
const candidates = candidatesById.get(line.id) ?? [];
|
||||
const decision = decisions.get(line.id) ?? null;
|
||||
// Sent to the AI but no answer came back for it: an error (retried
|
||||
// later), never a permanent no_match.
|
||||
const noAnswer = candidates.length > 0 && !aiFailed && !decision;
|
||||
try {
|
||||
const fresh = await pb.collection("quote_items").getOne<QuoteItem>(line.id);
|
||||
if (!isLineStillWritable(fresh) || fresh.description !== line.description) {
|
||||
|
|
@ -171,21 +181,26 @@ export async function fillQuote(
|
|||
}
|
||||
const patch = buildLinePatch({
|
||||
item: fresh,
|
||||
decision: decisions.get(line.id) ?? null,
|
||||
decision,
|
||||
candidates,
|
||||
minConfidence: settings.quote_assistant_min_confidence,
|
||||
model,
|
||||
now,
|
||||
status: candidates.length === 0 ? "no_candidates" : aiFailed ? "error" : undefined,
|
||||
status:
|
||||
candidates.length === 0 ? "no_candidates" : aiFailed || noAnswer ? "error" : undefined,
|
||||
});
|
||||
await pb.collection("quote_items").update(line.id, patch);
|
||||
const status = (patch.ai_fill as { status?: string } | undefined)?.status;
|
||||
if (status === "filled") result.filled += 1;
|
||||
else if (status === "no_match" || status === "no_candidates") result.unmatched += 1;
|
||||
if (noAnswer) unanswered += 1;
|
||||
} catch (e) {
|
||||
if (!result.error) result.error = `write failed for line ${line.id}: ${errMessage(e)}`;
|
||||
}
|
||||
}
|
||||
if (unanswered > 0 && !result.error) {
|
||||
result.error = `AI returned no decision for ${unanswered} line(s); they will be retried`;
|
||||
}
|
||||
return result;
|
||||
});
|
||||
}
|
||||
|
|
@ -200,7 +215,11 @@ type TickLine = QuoteItem & { expand?: { quote?: Quote } };
|
|||
*/
|
||||
export async function runAssistantTick(
|
||||
deps: AssistantDeps = {},
|
||||
): Promise<{ ran: boolean; reason?: "disabled" | "not_due"; summary?: QuoteAssistantRunSummary }> {
|
||||
): Promise<{
|
||||
ran: boolean;
|
||||
reason?: "disabled" | "not_due" | "running";
|
||||
summary?: QuoteAssistantRunSummary;
|
||||
}> {
|
||||
const nowFn = deps.now ?? (() => new Date());
|
||||
|
||||
clearAiSettingsCache();
|
||||
|
|
@ -216,6 +235,16 @@ export async function runAssistantTick(
|
|||
return { ran: false, reason: "not_due" };
|
||||
}
|
||||
|
||||
// In-progress lease: a run stamps last_run_at when it starts and writes
|
||||
// its summary when it ends. A start newer than the last summary means a
|
||||
// run is still going (n8n may have timed out on it); don't overlap it
|
||||
// unless it has been silent for longer than the lease.
|
||||
const lastFinished = Date.parse(settings.quote_assistant_last_run_summary?.finished_at || "");
|
||||
const unfinished = Number.isNaN(lastFinished) || lastFinished < last;
|
||||
if (!Number.isNaN(last) && unfinished && now.getTime() - last < TICK_LEASE_MS) {
|
||||
return { ran: false, reason: "running" };
|
||||
}
|
||||
|
||||
// Overlap guard: stamp the run start before doing any work.
|
||||
const startedAt = now.toISOString();
|
||||
await recordQuoteAssistantRun({ last_run_at: startedAt });
|
||||
|
|
@ -247,12 +276,18 @@ export async function runAssistantTick(
|
|||
if (!isLineEligible(item, { now })) continue;
|
||||
const group = perQuote.get(item.quote);
|
||||
if (!group && perQuote.size >= TICK_MAX_QUOTES) continue;
|
||||
// fillQuote won't take more than this; leave the budget to others.
|
||||
if (group && group.length >= QUOTE_MAX_LINES) continue;
|
||||
if (group) group.push(item);
|
||||
else perQuote.set(item.quote, [item]);
|
||||
taken += 1;
|
||||
}
|
||||
|
||||
const startMs = now.getTime();
|
||||
for (const [quoteId, group] of perQuote) {
|
||||
// Time budget: never start another quote late in the run, so a run
|
||||
// finishes inside n8n's request timeout. Leftovers go next run.
|
||||
if (nowFn().getTime() - startMs >= TICK_TIME_BUDGET_MS) break;
|
||||
const label = group[0].expand?.quote?.quote_number || quoteId;
|
||||
summary.quotes_seen += 1;
|
||||
try {
|
||||
|
|
|
|||
Loading…
Reference in a new issue