mirror of
https://github.com/rowboatlabs/rowboat.git
synced 2026-07-12 21:02:17 +02:00
move local-model concurrency to a turn-level defer lock
Review feedback: drop the model-layer scheduler (per-call-site interactive/classifier/background priorities, local-provider hostname heuristic) in favor of app-level orchestration: - deferBackgroundTasks flag in models.json, surfaced as a settings toggle; auto-enabled once (UI logic) when the user connects Ollama - ChatActivity counter marked by both chat runtimes (sessions layer and legacy AgentRuntime.trigger) - startWhenPossible/runWhenPossible wrappers around the headless agent runner; all background invocations (knowledge pipeline, live notes, background tasks, scheduled + prebuilt agents) go through them and wait for chat-idle when the flag is set createLanguageModel keeps only the Ollama context-window middleware; the LM Studio capability probe now keys off the provider flavor instead of hostname sniffing. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Arjun
parent
c1fd4e6221
commit
fd49334c33
33 changed files with 373 additions and 541 deletions
|
|
@ -364,6 +364,11 @@ function ModelSettings({ dialogOpen, rowboatConnected = false }: { dialogOpen: b
|
|||
const [testState, setTestState] = useState<{ status: "idle" | "testing" | "success" | "error"; error?: string }>({ status: "idle" })
|
||||
const [configLoading, setConfigLoading] = useState(true)
|
||||
const [showMoreProviders, setShowMoreProviders] = useState(false)
|
||||
// "Defer background tasks while a chat is running" — a top-level
|
||||
// models.json flag. deferExplicit tracks whether the user (or the Ollama
|
||||
// auto-enable) has ever set it, so we only auto-enable once.
|
||||
const [deferBackgroundTasks, setDeferBackgroundTasks] = useState(false)
|
||||
const [deferExplicit, setDeferExplicit] = useState(false)
|
||||
|
||||
const activeConfig = providerConfigs[provider]
|
||||
const showApiKey = provider === "openai" || provider === "anthropic" || provider === "google" || provider === "openrouter" || provider === "aigateway" || provider === "openai-compatible"
|
||||
|
|
@ -406,6 +411,8 @@ function ModelSettings({ dialogOpen, rowboatConnected = false }: { dialogOpen: b
|
|||
path: "config/models.json",
|
||||
})
|
||||
const parsed = JSON.parse(result.data)
|
||||
setDeferBackgroundTasks(parsed?.deferBackgroundTasks === true)
|
||||
setDeferExplicit(typeof parsed?.deferBackgroundTasks === "boolean")
|
||||
if (parsed?.provider?.flavor && parsed?.model) {
|
||||
const flavor = parsed.provider.flavor as LlmProviderFlavor
|
||||
setProvider(flavor)
|
||||
|
|
@ -462,6 +469,17 @@ function ModelSettings({ dialogOpen, rowboatConnected = false }: { dialogOpen: b
|
|||
loadCurrentConfig()
|
||||
}, [dialogOpen])
|
||||
|
||||
const handleDeferToggle = useCallback(async (value: boolean) => {
|
||||
setDeferBackgroundTasks(value)
|
||||
setDeferExplicit(true)
|
||||
try {
|
||||
await window.ipc.invoke("models:updateConfig", { deferBackgroundTasks: value })
|
||||
window.dispatchEvent(new Event("models-config-changed"))
|
||||
} catch {
|
||||
toast.error("Failed to save setting")
|
||||
}
|
||||
}, [])
|
||||
|
||||
// Load models catalog
|
||||
useEffect(() => {
|
||||
if (!dialogOpen) return
|
||||
|
|
@ -532,6 +550,12 @@ function ModelSettings({ dialogOpen, rowboatConnected = false }: { dialogOpen: b
|
|||
setDefaultProvider(provider)
|
||||
setTestState({ status: "success" })
|
||||
window.dispatchEvent(new Event('models-config-changed'))
|
||||
// Local models compete with background agents for the same hardware:
|
||||
// when the user connects Ollama and has never touched the defer
|
||||
// flag, enable it for them (they can switch it off below).
|
||||
if (provider === "ollama" && !deferExplicit && !deferBackgroundTasks) {
|
||||
void handleDeferToggle(true)
|
||||
}
|
||||
// Capability probe caveats (local models): saved, but the user should
|
||||
// know when the model can't do tools or has a too-small context.
|
||||
const warnings: string[] = result.warnings ?? []
|
||||
|
|
@ -551,7 +575,7 @@ function ModelSettings({ dialogOpen, rowboatConnected = false }: { dialogOpen: b
|
|||
setTestState({ status: "error", error: "Connection test failed" })
|
||||
toast.error("Connection test failed")
|
||||
}
|
||||
}, [canTest, provider, activeConfig, rowboatConnected])
|
||||
}, [canTest, provider, activeConfig, rowboatConnected, deferExplicit, deferBackgroundTasks, handleDeferToggle])
|
||||
|
||||
const handleSetDefault = useCallback(async (prov: LlmProviderFlavor) => {
|
||||
const config = providerConfigs[prov]
|
||||
|
|
@ -938,6 +962,17 @@ function ModelSettings({ dialogOpen, rowboatConnected = false }: { dialogOpen: b
|
|||
</div>
|
||||
)}
|
||||
|
||||
{/* Defer background tasks while chatting */}
|
||||
<div className="flex items-center justify-between gap-4 rounded-md border px-3 py-2.5">
|
||||
<div className="min-w-0">
|
||||
<div className="text-sm font-medium">Defer background tasks while chatting</div>
|
||||
<div className="text-xs text-muted-foreground mt-0.5">
|
||||
Background agents (knowledge sync, live notes, tasks) wait until your chat finishes. Recommended for local models.
|
||||
</div>
|
||||
</div>
|
||||
<Switch checked={deferBackgroundTasks} onCheckedChange={handleDeferToggle} />
|
||||
</div>
|
||||
|
||||
{/* Test & Save button */}
|
||||
<Button
|
||||
onClick={handleTestAndSave}
|
||||
|
|
|
|||
|
|
@ -4,7 +4,7 @@ import { IAgentScheduleRepo } from "./repo.js";
|
|||
import { IAgentScheduleStateRepo } from "./state-repo.js";
|
||||
import { AgentScheduleConfig, AgentScheduleEntry } from "@x/shared/dist/agent-schedule.js";
|
||||
import { AgentScheduleState, AgentScheduleStateEntry } from "@x/shared/dist/agent-schedule-state.js";
|
||||
import { startHeadlessAgent } from "../agents/headless-app.js";
|
||||
import { startWhenPossible } from "../agents/headless-app.js";
|
||||
import { withUseCase } from "../analytics/use_case.js";
|
||||
import z from "zod";
|
||||
|
||||
|
|
@ -165,7 +165,7 @@ async function runAgent(
|
|||
const startingMessage = entry.startingMessage ?? DEFAULT_STARTING_MESSAGE;
|
||||
const handle = await withUseCase(
|
||||
{ useCase: 'copilot_chat', subUseCase: 'scheduled' },
|
||||
() => startHeadlessAgent({
|
||||
() => startWhenPossible({
|
||||
agentId: agentName,
|
||||
message: startingMessage,
|
||||
signal: AbortSignal.timeout(TIMEOUT_MS),
|
||||
|
|
|
|||
|
|
@ -1,4 +1,6 @@
|
|||
import container from "../di/container.js";
|
||||
import { chatActivity } from "../application/lib/chat-activity.js";
|
||||
import { shouldDeferBackgroundTasks } from "../models/defaults.js";
|
||||
import {
|
||||
type HeadlessAgentHandle,
|
||||
type HeadlessAgentOptions,
|
||||
|
|
@ -22,4 +24,30 @@ export function runHeadlessAgent(
|
|||
return runner().run(options);
|
||||
}
|
||||
|
||||
// When the user enabled "Defer background tasks while a chat is running"
|
||||
// (recommended for local models — a background run competes with the chat
|
||||
// for the same hardware), wait for the chat to go idle before starting.
|
||||
// Re-check after each wake: another chat turn may have started meanwhile.
|
||||
async function waitForChatIdleIfConfigured(): Promise<void> {
|
||||
while ((await shouldDeferBackgroundTasks()) && chatActivity.activeCount > 0) {
|
||||
await chatActivity.waitUntilIdle();
|
||||
}
|
||||
}
|
||||
|
||||
/** startHeadlessAgent for background work: honors the defer-while-chatting setting. */
|
||||
export async function startWhenPossible(
|
||||
options: HeadlessAgentOptions,
|
||||
): Promise<HeadlessAgentHandle> {
|
||||
await waitForChatIdleIfConfigured();
|
||||
return startHeadlessAgent(options);
|
||||
}
|
||||
|
||||
/** runHeadlessAgent for background work: honors the defer-while-chatting setting. */
|
||||
export async function runWhenPossible(
|
||||
options: HeadlessAgentOptions,
|
||||
): Promise<HeadlessAgentResult & { turnId: string }> {
|
||||
await waitForChatIdleIfConfigured();
|
||||
return runHeadlessAgent(options);
|
||||
}
|
||||
|
||||
export { toolInputPaths } from "./headless.js";
|
||||
|
|
|
|||
|
|
@ -20,6 +20,7 @@ import container from "../di/container.js";
|
|||
import { notifyIfEnabled } from "../application/notification/notifier.js";
|
||||
import { IModelConfigRepo } from "../models/repo.js";
|
||||
import { createLanguageModel } from "../models/models.js";
|
||||
import { chatActivity } from "../application/lib/chat-activity.js";
|
||||
import { resolveProviderConfig } from "../models/defaults.js";
|
||||
import { IAgentsRepo } from "./repo.js";
|
||||
import { IMonotonicallyIncreasingIdGenerator } from "../application/lib/id-gen.js";
|
||||
|
|
@ -496,6 +497,9 @@ export class AgentRuntime implements IAgentRuntime {
|
|||
return;
|
||||
}
|
||||
const signal = this.abortRegistry.createForRun(runId);
|
||||
// Legacy runs are user-facing chats: mark activity so background
|
||||
// agents can defer (see agents/headless-app.ts runWhenPossible).
|
||||
chatActivity.enter();
|
||||
try {
|
||||
await this.bus.publish({
|
||||
runId,
|
||||
|
|
@ -610,6 +614,7 @@ export class AgentRuntime implements IAgentRuntime {
|
|||
await this.runsRepo.appendEvents(runId, [errorEvent]);
|
||||
await this.bus.publish(errorEvent);
|
||||
} finally {
|
||||
chatActivity.exit();
|
||||
this.abortRegistry.cleanup(runId);
|
||||
await this.runsLock.release(runId);
|
||||
await this.bus.publish({
|
||||
|
|
@ -1365,8 +1370,7 @@ export async function* streamAgent({
|
|||
}
|
||||
const modelId = state.runModel;
|
||||
const providerConfig = await resolveProviderConfig(state.runProvider);
|
||||
// Legacy runs are user-facing chats: interactive priority on local models.
|
||||
const model = createLanguageModel(providerConfig, modelId, { priority: "interactive" });
|
||||
const model = createLanguageModel(providerConfig, modelId);
|
||||
logger.log(`using model: ${modelId} (provider: ${state.runProvider})`);
|
||||
|
||||
// Install use-case context for tool-internal LLM calls (e.g. parseFile)
|
||||
|
|
|
|||
|
|
@ -584,8 +584,7 @@ export const BuiltinTools: z.infer<typeof BuiltinToolsSchema> = {
|
|||
|
||||
const { model: modelId, provider: providerName } = await getDefaultModelAndProvider();
|
||||
const providerConfig = await resolveProviderConfig(providerName);
|
||||
// Runs as a tool inside a chat turn more often than not.
|
||||
const model = createLanguageModel(providerConfig, modelId, { priority: 'interactive' });
|
||||
const model = createLanguageModel(providerConfig, modelId);
|
||||
|
||||
const userPrompt = prompt || 'Convert this file to well-structured markdown.';
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,50 @@
|
|||
import { describe, expect, it } from "vitest";
|
||||
import { ChatActivity } from "./chat-activity.js";
|
||||
|
||||
const tick = () => new Promise<void>((r) => setTimeout(r, 0));
|
||||
|
||||
describe("ChatActivity", () => {
|
||||
it("resolves waiters immediately when idle", async () => {
|
||||
const activity = new ChatActivity();
|
||||
await activity.waitUntilIdle();
|
||||
expect(activity.activeCount).toBe(0);
|
||||
});
|
||||
|
||||
it("blocks waiters until every active chat exits", async () => {
|
||||
const activity = new ChatActivity();
|
||||
activity.enter();
|
||||
activity.enter();
|
||||
|
||||
let released = false;
|
||||
const waiting = activity.waitUntilIdle().then(() => {
|
||||
released = true;
|
||||
});
|
||||
|
||||
activity.exit();
|
||||
await tick();
|
||||
expect(released).toBe(false);
|
||||
|
||||
activity.exit();
|
||||
await waiting;
|
||||
expect(released).toBe(true);
|
||||
});
|
||||
|
||||
it("wakes all pending waiters at once", async () => {
|
||||
const activity = new ChatActivity();
|
||||
activity.enter();
|
||||
const results: number[] = [];
|
||||
const a = activity.waitUntilIdle().then(() => results.push(1));
|
||||
const b = activity.waitUntilIdle().then(() => results.push(2));
|
||||
activity.exit();
|
||||
await Promise.all([a, b]);
|
||||
expect(results.sort()).toEqual([1, 2]);
|
||||
});
|
||||
|
||||
it("tolerates unbalanced exits", () => {
|
||||
const activity = new ChatActivity();
|
||||
activity.exit();
|
||||
expect(activity.activeCount).toBe(0);
|
||||
activity.enter();
|
||||
expect(activity.activeCount).toBe(1);
|
||||
});
|
||||
});
|
||||
39
apps/x/packages/core/src/application/lib/chat-activity.ts
Normal file
39
apps/x/packages/core/src/application/lib/chat-activity.ts
Normal file
|
|
@ -0,0 +1,39 @@
|
|||
// Tracks whether any user-facing chat turn is currently being processed.
|
||||
// Both chat runtimes mark their turns here (sessions/sessions.ts for the
|
||||
// turns runtime, agents/runtime.ts trigger() for legacy runs). Background
|
||||
// agent invocations consult it via startWhenPossible/runWhenPossible in
|
||||
// agents/headless-app.ts when the user enabled "Defer background tasks
|
||||
// while a chat is running" — useful on local models, where a background run
|
||||
// competes with the chat for the same hardware.
|
||||
export class ChatActivity {
|
||||
private active = 0;
|
||||
private waiters: Array<() => void> = [];
|
||||
|
||||
enter(): void {
|
||||
this.active++;
|
||||
}
|
||||
|
||||
exit(): void {
|
||||
this.active = Math.max(0, this.active - 1);
|
||||
if (this.active === 0) {
|
||||
const waiters = this.waiters.splice(0);
|
||||
for (const waiter of waiters) {
|
||||
waiter();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
get activeCount(): number {
|
||||
return this.active;
|
||||
}
|
||||
|
||||
/** Resolves immediately when no chat turn is running. */
|
||||
waitUntilIdle(): Promise<void> {
|
||||
if (this.active === 0) {
|
||||
return Promise.resolve();
|
||||
}
|
||||
return new Promise((resolve) => this.waiters.push(resolve));
|
||||
}
|
||||
}
|
||||
|
||||
export const chatActivity = new ChatActivity();
|
||||
|
|
@ -12,7 +12,7 @@ async function resolveRoutingModel() {
|
|||
const { model: modelId, provider } = await getBackgroundTaskAgentModel();
|
||||
const config = await resolveProviderConfig(provider);
|
||||
return {
|
||||
model: createLanguageModel(config, modelId, { priority: 'classifier' }),
|
||||
model: createLanguageModel(config, modelId),
|
||||
modelId,
|
||||
providerName: provider,
|
||||
};
|
||||
|
|
|
|||
|
|
@ -2,7 +2,7 @@ import type { BackgroundTask, BackgroundTaskTriggerType } from '@x/shared/dist/b
|
|||
import { PrefixLogger } from '@x/shared/dist/prefix-logger.js';
|
||||
import { fetchTask, patchTask, prependRunId } from './fileops.js';
|
||||
import { getBackgroundTaskAgentModel } from '../models/defaults.js';
|
||||
import { startHeadlessAgent } from '../agents/headless-app.js';
|
||||
import { startHeadlessAgent, startWhenPossible } from '../agents/headless-app.js';
|
||||
import { buildTriggerBlock } from '../agents/build-trigger-block.js';
|
||||
import { backgroundTaskBus } from './bus.js';
|
||||
import { withUseCase } from '../analytics/use_case.js';
|
||||
|
|
@ -143,13 +143,19 @@ export async function runBackgroundTask(
|
|||
const selection = await getBackgroundTaskAgentModel();
|
||||
const model = task.model || selection.model;
|
||||
const provider = task.provider ?? (task.model ? undefined : selection.provider);
|
||||
// Manual runs are user-requested (the Run button, or the copilot's
|
||||
// run-background-task-agent tool mid-chat) and must NOT wait for
|
||||
// chat-idle: the requesting chat turn holds the chat-activity lock,
|
||||
// so deferring here would deadlock the turn. Only autonomous
|
||||
// triggers (cron/window/event) defer.
|
||||
const start = trigger === 'manual' ? startHeadlessAgent : startWhenPossible;
|
||||
// Establish the use-case context for the whole turn so every tool the
|
||||
// agent calls (notably notify-user) reads `background_task_agent` via
|
||||
// getCurrentUseCase(); the AsyncLocalStorage context set here flows
|
||||
// through the turn's async execution chain.
|
||||
const handle = await withUseCase(
|
||||
{ useCase: 'background_task_agent', subUseCase: trigger },
|
||||
() => startHeadlessAgent({
|
||||
() => start({
|
||||
agentId: 'background-task-agent',
|
||||
message: buildMessage(slug, task, trigger, context, codeProject),
|
||||
model,
|
||||
|
|
|
|||
|
|
@ -2,7 +2,7 @@ import fs from 'fs';
|
|||
import path from 'path';
|
||||
import { google } from 'googleapis';
|
||||
import { WorkDir } from '../config/config.js';
|
||||
import { runHeadlessAgent } from '../agents/headless-app.js';
|
||||
import { runWhenPossible } from '../agents/headless-app.js';
|
||||
import { getKgModel } from '../models/defaults.js';
|
||||
import { getErrorDetails } from '../agents/utils.js';
|
||||
import { serviceLogger } from '../services/service_logger.js';
|
||||
|
|
@ -281,7 +281,7 @@ async function processAgentNotes(): Promise<void> {
|
|||
const timestamp = new Date().toISOString();
|
||||
const message = `Current timestamp: ${timestamp}\n\nProcess the following source material and update the Agent Notes folder accordingly.\n\n${messageParts.join('\n\n')}`;
|
||||
|
||||
await runHeadlessAgent({
|
||||
await runWhenPossible({
|
||||
agentId: AGENT_ID,
|
||||
message,
|
||||
...(await getKgModel()),
|
||||
|
|
|
|||
|
|
@ -2,7 +2,7 @@ import fs from 'fs';
|
|||
import path from 'path';
|
||||
import { WorkDir } from '../config/config.js';
|
||||
import { getKgModel } from '../models/defaults.js';
|
||||
import { runHeadlessAgent, toolInputPaths } from '../agents/headless-app.js';
|
||||
import { runWhenPossible, toolInputPaths } from '../agents/headless-app.js';
|
||||
import { getErrorDetails } from '../agents/utils.js';
|
||||
import { serviceLogger, type ServiceRunContext } from '../services/service_logger.js';
|
||||
import {
|
||||
|
|
@ -385,7 +385,7 @@ async function createNotesFromBatch(
|
|||
message += `(3) No placeholder text ("Unknown"/"-") and no links between entities that didn't co-occur in one source file.\n`;
|
||||
}
|
||||
|
||||
const { turnId, state } = await runHeadlessAgent({
|
||||
const { turnId, state } = await runWhenPossible({
|
||||
agentId: NOTE_CREATION_AGENT,
|
||||
message,
|
||||
...(await getKgModel()),
|
||||
|
|
@ -943,7 +943,7 @@ export async function curateNotes(): Promise<void> {
|
|||
message += `Curate the following knowledge note per your instructions. Rewrite it in place with a single file-writeText to the SAME path.\n\n`;
|
||||
message += `**Note path:** ${relPath}\n\n`;
|
||||
message += `**Current content:**\n\n${content}\n`;
|
||||
await runHeadlessAgent({
|
||||
await runWhenPossible({
|
||||
agentId: CURATION_AGENT,
|
||||
message,
|
||||
...(await getKgModel()),
|
||||
|
|
|
|||
|
|
@ -247,7 +247,7 @@ export async function classifyThread(
|
|||
|
||||
const { model: modelId, provider } = await getKgModel();
|
||||
const config = await resolveProviderConfig(provider);
|
||||
const model = createLanguageModel(config, modelId, { priority: 'background' });
|
||||
const model = createLanguageModel(config, modelId);
|
||||
|
||||
let systemPrompt = options.skipDraft
|
||||
? `${SYSTEM_PROMPT}\n\n# Skip the draft\n\nThe user already has their own draft in progress for this thread — DO NOT generate a draftResponse. Always omit the draftResponse field.`
|
||||
|
|
|
|||
|
|
@ -147,7 +147,7 @@ export async function maybeDistillImportanceRules(): Promise<void> {
|
|||
try {
|
||||
const { model: modelId, provider } = await getKgModel();
|
||||
const config = await resolveProviderConfig(provider);
|
||||
const model = createLanguageModel(config, modelId, { priority: 'background' });
|
||||
const model = createLanguageModel(config, modelId);
|
||||
|
||||
const correctionLines = fb.corrections.map(c =>
|
||||
`- From: ${c.from} | Subject: "${c.subject}" | classifier said ${c.agentVerdict}, user corrected to ${c.userVerdict}`
|
||||
|
|
|
|||
|
|
@ -3,7 +3,7 @@ import path from 'path';
|
|||
import { CronExpressionParser } from 'cron-parser';
|
||||
import { generateText } from 'ai';
|
||||
import { WorkDir } from '../config/config.js';
|
||||
import { runHeadlessAgent } from '../agents/headless-app.js';
|
||||
import { runWhenPossible } from '../agents/headless-app.js';
|
||||
import { getKgModel } from '../models/defaults.js';
|
||||
import container from '../di/container.js';
|
||||
import type { IModelConfigRepo } from '../models/repo.js';
|
||||
|
|
@ -480,7 +480,7 @@ async function processInlineTasks(): Promise<void> {
|
|||
'```',
|
||||
].join('\n');
|
||||
|
||||
const { summary: result } = await runHeadlessAgent({
|
||||
const { summary: result } = await runWhenPossible({
|
||||
agentId: INLINE_TASK_AGENT,
|
||||
message,
|
||||
...(await getKgModel()),
|
||||
|
|
@ -559,7 +559,7 @@ export async function processRowboatInstruction(
|
|||
'```',
|
||||
].join('\n');
|
||||
|
||||
const { summary: rawResponse } = await runHeadlessAgent({
|
||||
const { summary: rawResponse } = await runWhenPossible({
|
||||
agentId: INLINE_TASK_AGENT,
|
||||
message,
|
||||
...(await getKgModel()),
|
||||
|
|
@ -613,7 +613,7 @@ export async function processRowboatInstruction(
|
|||
export async function classifySchedule(instruction: string): Promise<InlineTaskSchedule | null> {
|
||||
const repo = container.resolve<IModelConfigRepo>('modelConfigRepo');
|
||||
const config = await repo.getConfig();
|
||||
const model = createLanguageModel(config.provider, config.model, { priority: 'classifier' });
|
||||
const model = createLanguageModel(config.provider, config.model);
|
||||
|
||||
const now = new Date();
|
||||
const defaultEnd = new Date(now.getTime() + 7 * 24 * 60 * 60 * 1000);
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
import fs from 'fs';
|
||||
import path from 'path';
|
||||
import { WorkDir } from '../config/config.js';
|
||||
import { runHeadlessAgent, toolInputPaths } from '../agents/headless-app.js';
|
||||
import { runWhenPossible, toolInputPaths } from '../agents/headless-app.js';
|
||||
import { getKgModel } from '../models/defaults.js';
|
||||
import { getErrorDetails } from '../agents/utils.js';
|
||||
import { serviceLogger } from '../services/service_logger.js';
|
||||
|
|
@ -84,7 +84,7 @@ async function labelEmailBatch(
|
|||
message += `\n\n---\n\n`;
|
||||
}
|
||||
|
||||
const { turnId, state } = await runHeadlessAgent({
|
||||
const { turnId, state } = await runWhenPossible({
|
||||
agentId: LABELING_AGENT,
|
||||
message,
|
||||
...(await getKgModel()),
|
||||
|
|
|
|||
|
|
@ -10,7 +10,7 @@ async function resolveRoutingModel() {
|
|||
const { model: modelId, provider } = await getLiveNoteAgentModel();
|
||||
const config = await resolveProviderConfig(provider);
|
||||
return {
|
||||
model: createLanguageModel(config, modelId, { priority: 'classifier' }),
|
||||
model: createLanguageModel(config, modelId),
|
||||
modelId,
|
||||
providerName: provider,
|
||||
};
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
import type { LiveNote, LiveNoteTriggerType } from '@x/shared/dist/live-note.js';
|
||||
import { fetchLiveNote, patchLiveNote, readNoteBody } from './fileops.js';
|
||||
import { getLiveNoteAgentModel } from '../../models/defaults.js';
|
||||
import { startHeadlessAgent } from '../../agents/headless-app.js';
|
||||
import { startHeadlessAgent, startWhenPossible } from '../../agents/headless-app.js';
|
||||
import { withUseCase } from '../../analytics/use_case.js';
|
||||
import { buildTriggerBlock } from '../../agents/build-trigger-block.js';
|
||||
import { liveNoteBus } from './bus.js';
|
||||
|
|
@ -114,12 +114,18 @@ export async function runLiveNoteAgent(
|
|||
const selection = await getLiveNoteAgentModel();
|
||||
const model = live.model ?? selection.model;
|
||||
const provider = live.provider ?? (live.model ? undefined : selection.provider);
|
||||
// Manual runs are user-requested (the Run button, or the copilot's
|
||||
// run-live-note-agent tool mid-chat) and must NOT wait for chat-idle:
|
||||
// the requesting chat turn holds the chat-activity lock, so deferring
|
||||
// here would deadlock the turn. Only autonomous triggers
|
||||
// (cron/window/event) defer.
|
||||
const start = trigger === 'manual' ? startHeadlessAgent : startWhenPossible;
|
||||
// The use-case context propagates to every tool the agent calls; the
|
||||
// granular trigger doubles as the sub-use-case (manual / cron /
|
||||
// window / event) so dashboards can break down what woke the agent.
|
||||
const handle = await withUseCase(
|
||||
{ useCase: 'live_note_agent', subUseCase: trigger },
|
||||
() => startHeadlessAgent({
|
||||
() => start({
|
||||
agentId: 'live-note-agent',
|
||||
message: buildMessage(filePath, live, trigger, context),
|
||||
model,
|
||||
|
|
|
|||
|
|
@ -177,7 +177,7 @@ async function generateBrief(event: CalendarEvent, ctx: Awaited<ReturnType<typeo
|
|||
|
||||
const { model: modelId, provider: providerName } = await getMeetingNotesModel();
|
||||
const providerConfig = await resolveProviderConfig(providerName);
|
||||
const model = createLanguageModel(providerConfig, modelId, { priority: 'background' });
|
||||
const model = createLanguageModel(providerConfig, modelId);
|
||||
|
||||
const result = await withUseCase({ useCase: 'meeting_prep' }, () => generateText({
|
||||
model,
|
||||
|
|
|
|||
|
|
@ -139,7 +139,7 @@ function loadCalendarEventContext(calendarEventJson: string): string {
|
|||
export async function summarizeMeeting(transcript: string, meetingStartTime?: string, calendarEventJson?: string): Promise<string> {
|
||||
const { model: modelId, provider: providerName } = await getMeetingNotesModel();
|
||||
const providerConfig = await resolveProviderConfig(providerName);
|
||||
const model = createLanguageModel(providerConfig, modelId, { priority: 'background' });
|
||||
const model = createLanguageModel(providerConfig, modelId);
|
||||
|
||||
// If a specific calendar event was linked, use it directly.
|
||||
// Otherwise fall back to scanning events within ±3 hours.
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
import fs from 'fs';
|
||||
import path from 'path';
|
||||
import { WorkDir } from '../config/config.js';
|
||||
import { runHeadlessAgent, toolInputPaths } from '../agents/headless-app.js';
|
||||
import { runWhenPossible, toolInputPaths } from '../agents/headless-app.js';
|
||||
import { getKgModel } from '../models/defaults.js';
|
||||
import { getErrorDetails } from '../agents/utils.js';
|
||||
import { serviceLogger } from '../services/service_logger.js';
|
||||
|
|
@ -97,7 +97,7 @@ async function tagNoteBatch(
|
|||
message += `\n\n---\n\n`;
|
||||
}
|
||||
|
||||
const { turnId, state } = await runHeadlessAgent({
|
||||
const { turnId, state } = await runWhenPossible({
|
||||
agentId: NOTE_TAGGING_AGENT,
|
||||
message,
|
||||
...(await getKgModel()),
|
||||
|
|
|
|||
|
|
@ -54,6 +54,16 @@ export async function getDefaultModelAndProvider(): Promise<{ model: string; pro
|
|||
return { model: cfg.model, provider: cfg.provider.flavor };
|
||||
}
|
||||
|
||||
/**
|
||||
* "Defer background tasks while a chat is running" (settings checkbox,
|
||||
* models.json `deferBackgroundTasks`). Read at each background invocation so
|
||||
* toggling takes effect immediately.
|
||||
*/
|
||||
export async function shouldDeferBackgroundTasks(): Promise<boolean> {
|
||||
const cfg = await readConfig();
|
||||
return cfg?.deferBackgroundTasks === true;
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve a provider name (as stored on a run, an agent, or returned by
|
||||
* getDefaultModelAndProvider) into the full LlmProvider config that
|
||||
|
|
|
|||
|
|
@ -1,37 +1,5 @@
|
|||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import {
|
||||
LocalLlmScheduler,
|
||||
isLocalProvider,
|
||||
makeOllamaThinkFetch,
|
||||
resolveThinkValue,
|
||||
} from "./local.js";
|
||||
|
||||
function deferred() {
|
||||
let resolve!: () => void;
|
||||
const promise = new Promise<void>((r) => {
|
||||
resolve = r;
|
||||
});
|
||||
return { promise, resolve };
|
||||
}
|
||||
|
||||
const tick = () => new Promise<void>((r) => setTimeout(r, 0));
|
||||
|
||||
describe("isLocalProvider", () => {
|
||||
it("treats ollama as local", () => {
|
||||
expect(isLocalProvider({ flavor: "ollama" })).toBe(true);
|
||||
});
|
||||
|
||||
it("treats loopback openai-compatible endpoints as local", () => {
|
||||
expect(isLocalProvider({ flavor: "openai-compatible", baseURL: "http://localhost:1234/v1" })).toBe(true);
|
||||
expect(isLocalProvider({ flavor: "openai-compatible", baseURL: "http://127.0.0.1:8080/v1" })).toBe(true);
|
||||
});
|
||||
|
||||
it("treats remote openai-compatible endpoints and cloud flavors as non-local", () => {
|
||||
expect(isLocalProvider({ flavor: "openai-compatible", baseURL: "https://api.together.xyz/v1" })).toBe(false);
|
||||
expect(isLocalProvider({ flavor: "openai" })).toBe(false);
|
||||
expect(isLocalProvider({ flavor: "rowboat" })).toBe(false);
|
||||
});
|
||||
});
|
||||
import { makeOllamaThinkFetch, resolveThinkValue } from "./local.js";
|
||||
|
||||
describe("resolveThinkValue", () => {
|
||||
it("passes effort levels straight through for gpt-oss variants", () => {
|
||||
|
|
@ -111,92 +79,3 @@ describe("makeOllamaThinkFetch", () => {
|
|||
expect(calls.some((c) => c.url.endsWith("/api/tags"))).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
describe("LocalLlmScheduler", () => {
|
||||
it("serves waiters by priority, FIFO within a priority", async () => {
|
||||
const scheduler = new LocalLlmScheduler(1);
|
||||
const order: string[] = [];
|
||||
const first = deferred();
|
||||
|
||||
const running = scheduler.run("background", undefined, async () => {
|
||||
await first.promise;
|
||||
order.push("initial");
|
||||
});
|
||||
await tick();
|
||||
|
||||
const bg1 = scheduler.run("background", undefined, async () => {
|
||||
order.push("bg1");
|
||||
});
|
||||
const bg2 = scheduler.run("background", undefined, async () => {
|
||||
order.push("bg2");
|
||||
});
|
||||
const chat = scheduler.run("interactive", undefined, async () => {
|
||||
order.push("chat");
|
||||
});
|
||||
const classifier = scheduler.run("classifier", undefined, async () => {
|
||||
order.push("classifier");
|
||||
});
|
||||
await tick();
|
||||
|
||||
first.resolve();
|
||||
await Promise.all([running, bg1, bg2, chat, classifier]);
|
||||
expect(order).toEqual(["initial", "chat", "classifier", "bg1", "bg2"]);
|
||||
});
|
||||
|
||||
it("releases the slot when a task throws", async () => {
|
||||
const scheduler = new LocalLlmScheduler(1);
|
||||
await expect(
|
||||
scheduler.run("interactive", undefined, async () => {
|
||||
throw new Error("boom");
|
||||
}),
|
||||
).rejects.toThrow("boom");
|
||||
// Slot must be free again.
|
||||
await scheduler.run("interactive", undefined, async () => undefined);
|
||||
});
|
||||
|
||||
it("rejects queued waiters whose signal aborts, without leaking the slot", async () => {
|
||||
const scheduler = new LocalLlmScheduler(1);
|
||||
const gate = deferred();
|
||||
const running = scheduler.run("background", undefined, () => gate.promise);
|
||||
await tick();
|
||||
|
||||
const controller = new AbortController();
|
||||
const waiting = scheduler.acquire("interactive", controller.signal);
|
||||
controller.abort();
|
||||
await expect(waiting).rejects.toThrow(/abort/i);
|
||||
|
||||
gate.resolve();
|
||||
await running;
|
||||
// Queue is clean: a fresh acquire succeeds immediately.
|
||||
const release = await scheduler.acquire("background");
|
||||
release();
|
||||
});
|
||||
|
||||
it("rejects immediately when acquiring with an already-aborted signal", async () => {
|
||||
const scheduler = new LocalLlmScheduler(1);
|
||||
const controller = new AbortController();
|
||||
controller.abort();
|
||||
await expect(scheduler.acquire("interactive", controller.signal)).rejects.toThrow(/abort/i);
|
||||
});
|
||||
|
||||
it("allows up to maxConcurrent tasks at once", async () => {
|
||||
const scheduler = new LocalLlmScheduler(2);
|
||||
const gateA = deferred();
|
||||
const gateB = deferred();
|
||||
let active = 0;
|
||||
let peak = 0;
|
||||
const track = async (gate: Promise<void>) => {
|
||||
active++;
|
||||
peak = Math.max(peak, active);
|
||||
await gate;
|
||||
active--;
|
||||
};
|
||||
const a = scheduler.run("background", undefined, () => track(gateA.promise));
|
||||
const b = scheduler.run("background", undefined, () => track(gateB.promise));
|
||||
await tick();
|
||||
expect(peak).toBe(2);
|
||||
gateA.resolve();
|
||||
gateB.resolve();
|
||||
await Promise.all([a, b]);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -1,9 +1,6 @@
|
|||
import { wrapLanguageModel, type LanguageModel } from "ai";
|
||||
import type { z } from "zod";
|
||||
import type { LlmProvider } from "@x/shared/dist/models.js";
|
||||
import { PrefixLogger } from "@x/shared";
|
||||
|
||||
const log = new PrefixLogger("LocalLlm");
|
||||
|
||||
// Ollama's server-side default context window (~4k tokens) is far below what
|
||||
// Rowboat's agents need (the copilot's system prompt + tool schemas alone are
|
||||
|
|
@ -13,218 +10,6 @@ const log = new PrefixLogger("LocalLlm");
|
|||
// `contextLength` in models.json.
|
||||
export const DEFAULT_OLLAMA_CONTEXT_LENGTH = 32768;
|
||||
|
||||
const LOOPBACK_HOSTS = new Set(["localhost", "127.0.0.1", "0.0.0.0", "::1", "[::1]"]);
|
||||
|
||||
/**
|
||||
* Whether requests to this provider are served by a model running on the
|
||||
* user's own machine (and therefore need scheduling: local runtimes process
|
||||
* requests mostly serially, so background work must not starve chat).
|
||||
*/
|
||||
export function isLocalProvider(config: z.infer<typeof LlmProvider>): boolean {
|
||||
if (config.flavor === "ollama") {
|
||||
return true;
|
||||
}
|
||||
if (config.flavor === "openai-compatible") {
|
||||
// LM Studio, llama.cpp server, vLLM on the same machine, etc.
|
||||
try {
|
||||
const host = new URL(config.baseURL ?? "").hostname;
|
||||
return LOOPBACK_HOSTS.has(host) || host.endsWith(".local");
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
export type LlmPriority = "interactive" | "classifier" | "background";
|
||||
|
||||
const PRIORITY_ORDER: Record<LlmPriority, number> = {
|
||||
interactive: 0,
|
||||
classifier: 1,
|
||||
background: 2,
|
||||
};
|
||||
|
||||
interface Waiter {
|
||||
priority: LlmPriority;
|
||||
seq: number;
|
||||
resolve: (release: () => void) => void;
|
||||
reject: (error: Error) => void;
|
||||
signal?: AbortSignal;
|
||||
onAbort?: () => void;
|
||||
}
|
||||
|
||||
function abortError(): Error {
|
||||
const error = new Error("local LLM slot acquisition aborted");
|
||||
error.name = "AbortError";
|
||||
return error;
|
||||
}
|
||||
|
||||
/**
|
||||
* Serializes access to local LLM runtimes. One slot by default: local servers
|
||||
* effectively process one request at a time, and interleaving requests
|
||||
* destroys their KV-cache reuse. Waiters are served strictly by priority
|
||||
* (interactive chat > lightweight classifiers > background knowledge
|
||||
* pipeline), FIFO within a priority — so a queued email-labeling job can
|
||||
* never delay the user's chat by more than the one request already running.
|
||||
*/
|
||||
export class LocalLlmScheduler {
|
||||
private active = 0;
|
||||
private seq = 0;
|
||||
private readonly waiting: Waiter[] = [];
|
||||
|
||||
constructor(private readonly maxConcurrent = 1) {}
|
||||
|
||||
async acquire(priority: LlmPriority, signal?: AbortSignal): Promise<() => void> {
|
||||
if (signal?.aborted) {
|
||||
throw abortError();
|
||||
}
|
||||
if (this.active < this.maxConcurrent) {
|
||||
this.active++;
|
||||
return this.makeRelease();
|
||||
}
|
||||
return new Promise<() => void>((resolve, reject) => {
|
||||
const waiter: Waiter = { priority, seq: this.seq++, resolve, reject, signal };
|
||||
if (signal) {
|
||||
waiter.onAbort = () => {
|
||||
const index = this.waiting.indexOf(waiter);
|
||||
if (index >= 0) {
|
||||
this.waiting.splice(index, 1);
|
||||
reject(abortError());
|
||||
}
|
||||
};
|
||||
signal.addEventListener("abort", waiter.onAbort, { once: true });
|
||||
}
|
||||
this.waiting.push(waiter);
|
||||
if (this.waiting.length === 1 || this.waiting.length % 10 === 0) {
|
||||
log.log(`queueing ${priority} request (${this.waiting.length} waiting, ${this.active} active)`);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/** Run `fn` while holding a slot. */
|
||||
async run<T>(priority: LlmPriority, signal: AbortSignal | undefined, fn: () => PromiseLike<T>): Promise<T> {
|
||||
const release = await this.acquire(priority, signal);
|
||||
try {
|
||||
return await fn();
|
||||
} finally {
|
||||
release();
|
||||
}
|
||||
}
|
||||
|
||||
get queueDepth(): number {
|
||||
return this.waiting.length;
|
||||
}
|
||||
|
||||
private makeRelease(): () => void {
|
||||
let released = false;
|
||||
return () => {
|
||||
if (released) {
|
||||
return;
|
||||
}
|
||||
released = true;
|
||||
this.active--;
|
||||
this.dequeue();
|
||||
};
|
||||
}
|
||||
|
||||
private dequeue(): void {
|
||||
while (this.active < this.maxConcurrent && this.waiting.length > 0) {
|
||||
let best = 0;
|
||||
for (let i = 1; i < this.waiting.length; i++) {
|
||||
const a = this.waiting[i];
|
||||
const b = this.waiting[best];
|
||||
if (
|
||||
PRIORITY_ORDER[a.priority] < PRIORITY_ORDER[b.priority] ||
|
||||
(PRIORITY_ORDER[a.priority] === PRIORITY_ORDER[b.priority] && a.seq < b.seq)
|
||||
) {
|
||||
best = i;
|
||||
}
|
||||
}
|
||||
const [waiter] = this.waiting.splice(best, 1);
|
||||
if (waiter.signal && waiter.onAbort) {
|
||||
waiter.signal.removeEventListener("abort", waiter.onAbort);
|
||||
}
|
||||
this.active++;
|
||||
waiter.resolve(this.makeRelease());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function envConcurrency(): number {
|
||||
const raw = process.env.ROWBOAT_LOCAL_LLM_CONCURRENCY;
|
||||
const parsed = raw ? Number.parseInt(raw, 10) : NaN;
|
||||
return Number.isFinite(parsed) && parsed > 0 ? parsed : 1;
|
||||
}
|
||||
|
||||
// One queue for all local providers: users run one local server, and even
|
||||
// with several, the machine's compute is the shared resource.
|
||||
export const localLlmScheduler = new LocalLlmScheduler(envConcurrency());
|
||||
|
||||
/**
|
||||
* Wrap a language model so every call requests an explicit context window
|
||||
* from Ollama (merged under the caller's providerOptions — an explicit
|
||||
* caller value wins) and, when `priority` is set and the provider is local,
|
||||
* goes through the shared scheduler.
|
||||
*/
|
||||
export function applyLocalModelSettings(
|
||||
model: LanguageModel,
|
||||
providerConfig: z.infer<typeof LlmProvider>,
|
||||
priority: LlmPriority | null,
|
||||
): LanguageModel {
|
||||
if (typeof model === "string") {
|
||||
// Bare model-id strings resolve through the global registry; local
|
||||
// providers never take this path.
|
||||
return model;
|
||||
}
|
||||
const local = isLocalProvider(providerConfig);
|
||||
const wantsNumCtx = providerConfig.flavor === "ollama";
|
||||
if (!wantsNumCtx && !(local && priority)) {
|
||||
return model;
|
||||
}
|
||||
const numCtx = providerConfig.contextLength ?? DEFAULT_OLLAMA_CONTEXT_LENGTH;
|
||||
const schedule = local && priority ? priority : null;
|
||||
return wrapLanguageModel({
|
||||
model,
|
||||
middleware: {
|
||||
...(wantsNumCtx
|
||||
? {
|
||||
transformParams: async ({ params }) => {
|
||||
const providerOptions = (params.providerOptions ?? {}) as Record<string, Record<string, unknown>>;
|
||||
const ollama = (providerOptions.ollama ?? {}) as Record<string, unknown>;
|
||||
const options = (ollama.options ?? {}) as Record<string, unknown>;
|
||||
return {
|
||||
...params,
|
||||
providerOptions: {
|
||||
...providerOptions,
|
||||
ollama: {
|
||||
...ollama,
|
||||
options: { num_ctx: numCtx, ...options },
|
||||
},
|
||||
},
|
||||
};
|
||||
},
|
||||
}
|
||||
: {}),
|
||||
...(schedule
|
||||
? {
|
||||
wrapGenerate: async ({ doGenerate, params }) =>
|
||||
localLlmScheduler.run(schedule, params.abortSignal, () => doGenerate()),
|
||||
wrapStream: async ({ doStream, params }) => {
|
||||
const release = await localLlmScheduler.acquire(schedule, params.abortSignal);
|
||||
try {
|
||||
const { stream, ...rest } = await doStream();
|
||||
return { ...rest, stream: releaseOnSettled(stream, release) };
|
||||
} catch (error) {
|
||||
release();
|
||||
throw error;
|
||||
}
|
||||
},
|
||||
}
|
||||
: {}),
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
export type ReasoningEffort = "low" | "medium" | "high";
|
||||
|
||||
// Local models default to snappy: gpt-oss at medium effort spends ~3x the
|
||||
|
|
@ -233,6 +18,43 @@ export type ReasoningEffort = "low" | "medium" | "high";
|
|||
// false, which thinking models like gpt-oss simply ignore).
|
||||
export const DEFAULT_OLLAMA_REASONING_EFFORT: ReasoningEffort = "low";
|
||||
|
||||
/**
|
||||
* Wrap a language model so every call requests an explicit context window
|
||||
* from Ollama (merged under the caller's providerOptions — an explicit
|
||||
* caller value wins). Non-Ollama providers pass through untouched.
|
||||
*/
|
||||
export function applyLocalModelSettings(
|
||||
model: LanguageModel,
|
||||
providerConfig: z.infer<typeof LlmProvider>,
|
||||
): LanguageModel {
|
||||
if (typeof model === "string" || providerConfig.flavor !== "ollama") {
|
||||
// Bare model-id strings resolve through the global registry; local
|
||||
// providers never take this path.
|
||||
return model;
|
||||
}
|
||||
const numCtx = providerConfig.contextLength ?? DEFAULT_OLLAMA_CONTEXT_LENGTH;
|
||||
return wrapLanguageModel({
|
||||
model,
|
||||
middleware: {
|
||||
transformParams: async ({ params }) => {
|
||||
const providerOptions = (params.providerOptions ?? {}) as Record<string, Record<string, unknown>>;
|
||||
const ollama = (providerOptions.ollama ?? {}) as Record<string, unknown>;
|
||||
const options = (ollama.options ?? {}) as Record<string, unknown>;
|
||||
return {
|
||||
...params,
|
||||
providerOptions: {
|
||||
...providerOptions,
|
||||
ollama: {
|
||||
...ollama,
|
||||
options: { num_ctx: numCtx, ...options },
|
||||
},
|
||||
},
|
||||
};
|
||||
},
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Map a configured effort to the Ollama `think` value for a given model.
|
||||
* - gpt-oss accepts effort levels directly ("low" | "medium" | "high").
|
||||
|
|
@ -316,29 +138,3 @@ export function makeOllamaThinkFetch(
|
|||
return fetch(input, init);
|
||||
};
|
||||
}
|
||||
|
||||
// Hold the slot until the provider stream drains, errors, or is cancelled —
|
||||
// a streaming response occupies the local server for its full duration.
|
||||
function releaseOnSettled<T>(stream: ReadableStream<T>, release: () => void): ReadableStream<T> {
|
||||
const reader = stream.getReader();
|
||||
return new ReadableStream<T>({
|
||||
async pull(controller) {
|
||||
try {
|
||||
const { done, value } = await reader.read();
|
||||
if (done) {
|
||||
release();
|
||||
controller.close();
|
||||
} else {
|
||||
controller.enqueue(value);
|
||||
}
|
||||
} catch (error) {
|
||||
release();
|
||||
controller.error(error);
|
||||
}
|
||||
},
|
||||
cancel(reason) {
|
||||
release();
|
||||
return reader.cancel(reason);
|
||||
},
|
||||
});
|
||||
}
|
||||
|
|
|
|||
|
|
@ -14,11 +14,9 @@ import { getChatModelIds } from "./models-dev.js";
|
|||
import { withUseCase } from "../analytics/use_case.js";
|
||||
import {
|
||||
applyLocalModelSettings,
|
||||
isLocalProvider,
|
||||
makeOllamaThinkFetch,
|
||||
DEFAULT_OLLAMA_CONTEXT_LENGTH,
|
||||
DEFAULT_OLLAMA_REASONING_EFFORT,
|
||||
type LlmPriority,
|
||||
} from "./local.js";
|
||||
|
||||
export const Provider = LlmProvider;
|
||||
|
|
@ -90,18 +88,14 @@ export function createProvider(config: z.infer<typeof Provider>): ProviderV2 {
|
|||
|
||||
/**
|
||||
* The one place model instances are created. Applies local-runtime settings
|
||||
* (explicit Ollama context window) and, when `priority` is given, routes the
|
||||
* call through the shared local scheduler so background work cannot starve
|
||||
* interactive chat. Pass `priority: null` when the caller does its own
|
||||
* scheduling (the turn runtime's model registry).
|
||||
* (explicit Ollama context window) on top of the raw provider model.
|
||||
*/
|
||||
export function createLanguageModel(
|
||||
providerConfig: z.infer<typeof Provider>,
|
||||
modelId: string,
|
||||
opts: { priority?: LlmPriority | null } = {},
|
||||
): LanguageModel {
|
||||
const model = createProvider(providerConfig).languageModel(modelId);
|
||||
return applyLocalModelSettings(model, providerConfig, opts.priority ?? null);
|
||||
return applyLocalModelSettings(model, providerConfig);
|
||||
}
|
||||
|
||||
export interface ModelCapabilities {
|
||||
|
|
@ -151,9 +145,10 @@ export async function probeModelCapabilities(
|
|||
}
|
||||
return result;
|
||||
}
|
||||
if (providerConfig.flavor === "openai-compatible" && isLocalProvider(providerConfig)) {
|
||||
if (providerConfig.flavor === "openai-compatible") {
|
||||
// LM Studio's enhanced REST API lives at /api/v0 on the same
|
||||
// origin as the OpenAI-compatible /v1 endpoint.
|
||||
// origin as the OpenAI-compatible /v1 endpoint. Non-LM Studio
|
||||
// endpoints just 404 here, which reports as "unknown".
|
||||
const origin = new URL(providerConfig.baseURL ?? "").origin;
|
||||
const res = await fetch(`${origin}/api/v0/models`, {
|
||||
headers: providerConfig.headers ?? {},
|
||||
|
|
@ -342,7 +337,7 @@ export async function generateOneShot(opts: GenerateTextOptions): Promise<Genera
|
|||
const modelId = opts.model || def.model;
|
||||
const providerName = opts.provider || def.provider;
|
||||
const providerConfig = await resolveProviderConfig(providerName);
|
||||
const languageModel = createLanguageModel(providerConfig, modelId, { priority: "interactive" });
|
||||
const languageModel = createLanguageModel(providerConfig, modelId);
|
||||
const result = await withUseCase(
|
||||
{ useCase: "copilot_chat", subUseCase: "email_compose" },
|
||||
() => generateText({
|
||||
|
|
|
|||
|
|
@ -10,7 +10,8 @@ export type ModelConfigPatch = {
|
|||
| "knowledgeGraphModel"
|
||||
| "meetingNotesModel"
|
||||
| "liveNoteAgentModel"
|
||||
| "autoPermissionDecisionModel"]?: z.infer<typeof ModelConfig>[K] | null;
|
||||
| "autoPermissionDecisionModel"
|
||||
| "deferBackgroundTasks"]?: z.infer<typeof ModelConfig>[K] | null;
|
||||
};
|
||||
|
||||
export interface IModelConfigRepo {
|
||||
|
|
@ -85,7 +86,7 @@ export class FSModelConfigRepo implements IModelConfigRepo {
|
|||
const raw = await fs.readFile(this.configPath, "utf8");
|
||||
const existing = JSON.parse(raw);
|
||||
existingSelections = Object.fromEntries(
|
||||
["defaultSelection", "knowledgeGraphModel", "meetingNotesModel", "liveNoteAgentModel", "autoPermissionDecisionModel"]
|
||||
["defaultSelection", "knowledgeGraphModel", "meetingNotesModel", "liveNoteAgentModel", "autoPermissionDecisionModel", "deferBackgroundTasks"]
|
||||
.filter((key) => existing[key] !== undefined && (config as Record<string, unknown>)[key] === undefined)
|
||||
.map((key) => [key, existing[key]]),
|
||||
);
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
import fs from 'fs';
|
||||
import path from 'path';
|
||||
import { WorkDir } from '../config/config.js';
|
||||
import { runHeadlessAgent } from '../agents/headless-app.js';
|
||||
import { runWhenPossible } from '../agents/headless-app.js';
|
||||
import { getKgModel } from '../models/defaults.js';
|
||||
import {
|
||||
loadConfig,
|
||||
|
|
@ -52,7 +52,7 @@ Process new items and use the user context above to identify yourself when draft
|
|||
// The agent file is expected to be in the agents directory with
|
||||
// the same name. Waits for the turn to settle (errors tolerated,
|
||||
// matching the old no-throwOnError wait).
|
||||
await runHeadlessAgent({
|
||||
await runWhenPossible({
|
||||
agentId: agentName,
|
||||
message,
|
||||
...(await getKgModel()),
|
||||
|
|
|
|||
|
|
@ -83,7 +83,7 @@ export async function classifyToolPermissions(input: {
|
|||
|
||||
const { model: modelId, provider: providerName } = await getAutoPermissionDecisionModel();
|
||||
const providerConfig = await resolveProviderConfig(providerName);
|
||||
const model = createLanguageModel(providerConfig, modelId, { priority: "classifier" });
|
||||
const model = createLanguageModel(providerConfig, modelId);
|
||||
|
||||
const result = await withUseCase(
|
||||
{
|
||||
|
|
|
|||
|
|
@ -17,6 +17,7 @@ import {
|
|||
reduceTurn,
|
||||
} from "@x/shared/dist/turns.js";
|
||||
import type { IMonotonicallyIncreasingIdGenerator } from "../application/lib/id-gen.js";
|
||||
import { chatActivity } from "../application/lib/chat-activity.js";
|
||||
import type {
|
||||
ITurnRuntime,
|
||||
Turn,
|
||||
|
|
@ -339,9 +340,17 @@ export class SessionsImpl implements ISessions {
|
|||
input?: TurnExternalInput,
|
||||
): TurnExecution {
|
||||
const controller = new AbortController();
|
||||
// A session turn is a user-facing chat: mark it active so background
|
||||
// agents can defer (see agents/headless-app.ts runWhenPossible).
|
||||
if (sessionId !== null) {
|
||||
chatActivity.enter();
|
||||
}
|
||||
const execution = this.turnRuntime.advanceTurn(turnId, input, {
|
||||
signal: controller.signal,
|
||||
});
|
||||
if (sessionId !== null) {
|
||||
void execution.outcome.catch(() => undefined).finally(() => chatActivity.exit());
|
||||
}
|
||||
this.active.set(turnId, { sessionId, controller, execution });
|
||||
|
||||
void (async () => {
|
||||
|
|
|
|||
|
|
@ -14,11 +14,7 @@ import type { JsonValue, ModelDescriptor, TurnUsage } from "@x/shared/dist/turns
|
|||
import { convertFromMessages } from "../../agents/runtime.js";
|
||||
import { resolveProviderConfig } from "../../models/defaults.js";
|
||||
import { createProvider } from "../../models/models.js";
|
||||
import {
|
||||
applyLocalModelSettings,
|
||||
isLocalProvider,
|
||||
localLlmScheduler,
|
||||
} from "../../models/local.js";
|
||||
import { applyLocalModelSettings } from "../../models/local.js";
|
||||
import type {
|
||||
IModelRegistry,
|
||||
LlmStreamEvent,
|
||||
|
|
@ -70,15 +66,11 @@ export class RealModelRegistry implements IModelRegistry {
|
|||
): Promise<ResolvedModel> {
|
||||
const providerConfig = await this.resolveProvider(descriptor.provider);
|
||||
const provider = this.createProviderImpl(providerConfig);
|
||||
// Local settings (Ollama context window) are applied here, but
|
||||
// scheduling happens per-step in run() where the turn's priority is
|
||||
// known — so priority stays null.
|
||||
// Local settings (Ollama context window) are applied here.
|
||||
const model = applyLocalModelSettings(
|
||||
provider.languageModel(descriptor.model),
|
||||
providerConfig,
|
||||
null,
|
||||
);
|
||||
const local = isLocalProvider(providerConfig);
|
||||
return {
|
||||
descriptor,
|
||||
// The structural -> wire conversion the app uses today: weaves
|
||||
|
|
@ -87,14 +79,13 @@ export class RealModelRegistry implements IModelRegistry {
|
|||
// per-message, so composed requests are byte-stable.
|
||||
encodeMessages: (messages) =>
|
||||
convertFromMessages(messages) as unknown as JsonValue[],
|
||||
stream: (request) => this.run(model, request, local),
|
||||
stream: (request) => this.run(model, request),
|
||||
};
|
||||
}
|
||||
|
||||
private async *run(
|
||||
model: LanguageModel,
|
||||
request: ModelStreamRequest,
|
||||
local: boolean,
|
||||
): AsyncGenerator<LlmStreamEvent, void, void> {
|
||||
const tools: ToolSet = {};
|
||||
for (const descriptor of request.tools) {
|
||||
|
|
@ -123,15 +114,6 @@ export class RealModelRegistry implements IModelRegistry {
|
|||
: {}),
|
||||
};
|
||||
|
||||
// One scheduler slot per model step: local runtimes serve requests
|
||||
// serially, and headless turns must not starve interactive chat.
|
||||
const release = local
|
||||
? await localLlmScheduler.acquire(
|
||||
request.priority ?? "interactive",
|
||||
request.signal,
|
||||
)
|
||||
: null;
|
||||
|
||||
const parts: Array<z.infer<typeof AssistantContentPart>> = [];
|
||||
let textBuffer = "";
|
||||
let reasoningBuffer = "";
|
||||
|
|
@ -139,112 +121,108 @@ export class RealModelRegistry implements IModelRegistry {
|
|||
let usage: z.infer<typeof TurnUsage> = {};
|
||||
let providerMetadata: JsonValue | undefined;
|
||||
|
||||
try {
|
||||
const result = this.invoke({
|
||||
model,
|
||||
system: request.systemPrompt,
|
||||
messages: request.messages as ModelMessage[],
|
||||
tools,
|
||||
abortSignal: request.signal,
|
||||
...generationParams,
|
||||
});
|
||||
const result = this.invoke({
|
||||
model,
|
||||
system: request.systemPrompt,
|
||||
messages: request.messages as ModelMessage[],
|
||||
tools,
|
||||
abortSignal: request.signal,
|
||||
...generationParams,
|
||||
});
|
||||
|
||||
for await (const raw of result.fullStream) {
|
||||
request.signal.throwIfAborted();
|
||||
const event = raw as {
|
||||
type: string;
|
||||
text?: string;
|
||||
toolCallId?: string;
|
||||
toolName?: string;
|
||||
input?: unknown;
|
||||
finishReason?: string;
|
||||
usage?: Record<string, number | undefined>;
|
||||
providerMetadata?: unknown;
|
||||
error?: unknown;
|
||||
};
|
||||
switch (event.type) {
|
||||
case "text-start":
|
||||
textBuffer = "";
|
||||
yield { type: "step_event", event: { type: "text_start" } };
|
||||
break;
|
||||
case "text-delta": {
|
||||
const delta = event.text ?? "";
|
||||
textBuffer += delta;
|
||||
const last = parts[parts.length - 1];
|
||||
if (last?.type === "text") {
|
||||
last.text += delta;
|
||||
} else {
|
||||
parts.push({ type: "text", text: delta });
|
||||
}
|
||||
yield { type: "text_delta", delta };
|
||||
break;
|
||||
for await (const raw of result.fullStream) {
|
||||
request.signal.throwIfAborted();
|
||||
const event = raw as {
|
||||
type: string;
|
||||
text?: string;
|
||||
toolCallId?: string;
|
||||
toolName?: string;
|
||||
input?: unknown;
|
||||
finishReason?: string;
|
||||
usage?: Record<string, number | undefined>;
|
||||
providerMetadata?: unknown;
|
||||
error?: unknown;
|
||||
};
|
||||
switch (event.type) {
|
||||
case "text-start":
|
||||
textBuffer = "";
|
||||
yield { type: "step_event", event: { type: "text_start" } };
|
||||
break;
|
||||
case "text-delta": {
|
||||
const delta = event.text ?? "";
|
||||
textBuffer += delta;
|
||||
const last = parts[parts.length - 1];
|
||||
if (last?.type === "text") {
|
||||
last.text += delta;
|
||||
} else {
|
||||
parts.push({ type: "text", text: delta });
|
||||
}
|
||||
case "text-end":
|
||||
yield {
|
||||
type: "step_event",
|
||||
event: { type: "text_end", text: textBuffer },
|
||||
};
|
||||
break;
|
||||
case "reasoning-start":
|
||||
reasoningBuffer = "";
|
||||
yield { type: "step_event", event: { type: "reasoning_start" } };
|
||||
break;
|
||||
case "reasoning-delta": {
|
||||
const delta = event.text ?? "";
|
||||
reasoningBuffer += delta;
|
||||
const last = parts[parts.length - 1];
|
||||
if (last?.type === "reasoning") {
|
||||
last.text += delta;
|
||||
} else {
|
||||
parts.push({ type: "reasoning", text: delta });
|
||||
}
|
||||
yield { type: "reasoning_delta", delta };
|
||||
break;
|
||||
}
|
||||
case "reasoning-end":
|
||||
yield {
|
||||
type: "step_event",
|
||||
event: { type: "reasoning_end", text: reasoningBuffer },
|
||||
};
|
||||
break;
|
||||
case "tool-call": {
|
||||
const toolCall = {
|
||||
type: "tool-call" as const,
|
||||
toolCallId: String(event.toolCallId),
|
||||
toolName: String(event.toolName),
|
||||
arguments: event.input,
|
||||
};
|
||||
parts.push(toolCall);
|
||||
yield { type: "step_event", event: { type: "tool_call", toolCall } };
|
||||
break;
|
||||
}
|
||||
case "finish-step": {
|
||||
finishReason = event.finishReason ?? "unknown";
|
||||
usage = mapUsage(event.usage);
|
||||
providerMetadata = toJsonValue(event.providerMetadata);
|
||||
yield {
|
||||
type: "step_event",
|
||||
event: {
|
||||
type: "finish_step",
|
||||
finishReason,
|
||||
usage,
|
||||
...(providerMetadata === undefined
|
||||
? {}
|
||||
: { providerMetadata }),
|
||||
},
|
||||
};
|
||||
break;
|
||||
}
|
||||
case "error":
|
||||
throw event.error instanceof Error
|
||||
? event.error
|
||||
: new Error(formatStreamError(event.error));
|
||||
default:
|
||||
break;
|
||||
yield { type: "text_delta", delta };
|
||||
break;
|
||||
}
|
||||
case "text-end":
|
||||
yield {
|
||||
type: "step_event",
|
||||
event: { type: "text_end", text: textBuffer },
|
||||
};
|
||||
break;
|
||||
case "reasoning-start":
|
||||
reasoningBuffer = "";
|
||||
yield { type: "step_event", event: { type: "reasoning_start" } };
|
||||
break;
|
||||
case "reasoning-delta": {
|
||||
const delta = event.text ?? "";
|
||||
reasoningBuffer += delta;
|
||||
const last = parts[parts.length - 1];
|
||||
if (last?.type === "reasoning") {
|
||||
last.text += delta;
|
||||
} else {
|
||||
parts.push({ type: "reasoning", text: delta });
|
||||
}
|
||||
yield { type: "reasoning_delta", delta };
|
||||
break;
|
||||
}
|
||||
case "reasoning-end":
|
||||
yield {
|
||||
type: "step_event",
|
||||
event: { type: "reasoning_end", text: reasoningBuffer },
|
||||
};
|
||||
break;
|
||||
case "tool-call": {
|
||||
const toolCall = {
|
||||
type: "tool-call" as const,
|
||||
toolCallId: String(event.toolCallId),
|
||||
toolName: String(event.toolName),
|
||||
arguments: event.input,
|
||||
};
|
||||
parts.push(toolCall);
|
||||
yield { type: "step_event", event: { type: "tool_call", toolCall } };
|
||||
break;
|
||||
}
|
||||
case "finish-step": {
|
||||
finishReason = event.finishReason ?? "unknown";
|
||||
usage = mapUsage(event.usage);
|
||||
providerMetadata = toJsonValue(event.providerMetadata);
|
||||
yield {
|
||||
type: "step_event",
|
||||
event: {
|
||||
type: "finish_step",
|
||||
finishReason,
|
||||
usage,
|
||||
...(providerMetadata === undefined
|
||||
? {}
|
||||
: { providerMetadata }),
|
||||
},
|
||||
};
|
||||
break;
|
||||
}
|
||||
case "error":
|
||||
throw event.error instanceof Error
|
||||
? event.error
|
||||
: new Error(formatStreamError(event.error));
|
||||
default:
|
||||
break;
|
||||
}
|
||||
} finally {
|
||||
release?.();
|
||||
}
|
||||
|
||||
yield {
|
||||
|
|
|
|||
|
|
@ -34,10 +34,6 @@ export interface ModelStreamRequest {
|
|||
tools: Array<z.infer<typeof ToolDescriptor>>;
|
||||
parameters: Record<string, JsonValue>;
|
||||
signal: AbortSignal;
|
||||
// Scheduling class for local-model queueing: interactive turns (a human
|
||||
// is watching) preempt queued headless/background work. Defaults to
|
||||
// "interactive" so an unset priority never slows a user down.
|
||||
priority?: "interactive" | "background";
|
||||
}
|
||||
|
||||
export interface ResolvedModel {
|
||||
|
|
|
|||
|
|
@ -951,11 +951,6 @@ class TurnAdvance {
|
|||
tools: composed.tools,
|
||||
parameters: composed.parameters,
|
||||
signal: this.signal,
|
||||
// Headless turns (no human waiting) yield the local-model
|
||||
// queue to interactive chat.
|
||||
priority: this.definition.config.humanAvailable
|
||||
? "interactive"
|
||||
: "background",
|
||||
})) {
|
||||
switch (event.type) {
|
||||
case "text_delta":
|
||||
|
|
|
|||
|
|
@ -623,6 +623,7 @@ const ipcSchemas = {
|
|||
meetingNotesModel: ModelOverride.nullable().optional(),
|
||||
liveNoteAgentModel: ModelOverride.nullable().optional(),
|
||||
autoPermissionDecisionModel: ModelOverride.nullable().optional(),
|
||||
deferBackgroundTasks: z.boolean().nullable().optional(),
|
||||
}),
|
||||
res: z.object({
|
||||
success: z.literal(true),
|
||||
|
|
|
|||
|
|
@ -37,6 +37,11 @@ export const LlmModelConfig = z.object({
|
|||
// the signed-in curated default and the legacy top-level provider/model
|
||||
// pair — this is what lets signed-in users default to a BYOK model.
|
||||
defaultSelection: ModelRef.optional(),
|
||||
// When true, background agent runs (knowledge pipeline, live notes,
|
||||
// background tasks) wait until no chat turn is running before starting.
|
||||
// Surfaced as a settings checkbox; recommended for local models, where a
|
||||
// background run competes with the chat for the same hardware.
|
||||
deferBackgroundTasks: z.boolean().optional(),
|
||||
providers: z.record(z.string(), z.object({
|
||||
apiKey: z.string().optional(),
|
||||
baseURL: z.string().optional(),
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue