diff --git a/.gitignore b/.gitignore index 6e605ca9..e5d5cc58 100644 --- a/.gitignore +++ b/.gitignore @@ -59,6 +59,8 @@ oldreference/ !/docs/architecture/*.md !/CLAUDE.md !/AGENTS.md +!/CONTEXT.md +!/docs/agents/*.md !/问题.md !/.planning/**/*.md !/src/main/prompt-builder/**/*.md diff --git a/CLAUDE.md b/CLAUDE.md index bf733b3c..e70ea12e 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -82,3 +82,17 @@ 2. **​结论先行​**: 直接给结论和修补方案。解释必须是短小精悍的中文大白话,不显示P0/P1等级。 3. **​表格化输出​**: 多数内容(尤其是评审、对比、多项任务)必须以Markdown表格输出。 4. **​强制收口​**: 结束对话必须明确告知用到的 skill。 + +## Agent skills + +### Issue tracker + +GitHub Issues(`gh` CLI)。详见 `docs/agents/issue-tracker.md`。 + +### Triage labels + +使用默认标签体系。详见 `docs/agents/triage-labels.md`。 + +### Domain docs + +Single-context 布局。详见 `docs/agents/domain.md`。 diff --git a/docs/agents/domain.md b/docs/agents/domain.md new file mode 100644 index 00000000..16af7994 --- /dev/null +++ b/docs/agents/domain.md @@ -0,0 +1,36 @@ +# Domain Docs + +How the engineering skills should consume this repo's domain documentation when exploring the codebase. + +## Before exploring, read these + +- **`CONTEXT.md`** at the repo root, or +- **`CONTEXT-MAP.md`** at the repo root if it exists — it points at one `CONTEXT.md` per context. Read each one relevant to the topic. +- **`docs/adr/`** — read ADRs that touch the area you're about to work in. In multi-context repos, also check `src//docs/adr/` for context-scoped decisions. + +If any of these files don't exist, **proceed silently**. Don't flag their absence; don't suggest creating them upfront. The producer skill (`/grill-with-docs`) creates them lazily when terms or decisions actually get resolved. + +## File structure + +Single-context repo (most repos): + +``` +/ +├── CONTEXT.md +├── docs/adr/ +│ ├── 0001-event-sourced-orders.md +│ └── 0002-postgres-for-write-model.md +└── src/ +``` + +## Use the glossary's vocabulary + +When your output names a domain concept (in an issue title, a refactor proposal, a hypothesis, a test name), use the term as defined in `CONTEXT.md`. Don't drift to synonyms the glossary explicitly avoids. + +If the concept you need isn't in the glossary yet, that's a signal — either you're inventing language the project doesn't use (reconsider) or there's a real gap (note it for `/grill-with-docs`). + +## Flag ADR conflicts + +If your output contradicts an existing ADR, surface it explicitly rather than silently overriding: + +> _Contradicts ADR-0007 (event-sourced orders) — but worth reopening because…_ diff --git a/docs/agents/issue-tracker.md b/docs/agents/issue-tracker.md new file mode 100644 index 00000000..cce77ecb --- /dev/null +++ b/docs/agents/issue-tracker.md @@ -0,0 +1,22 @@ +# Issue tracker: GitHub + +Issues and PRDs for this repo live as GitHub issues. Use the `gh` CLI for all operations. + +## Conventions + +- **Create an issue**: `gh issue create --title "..." --body "..."`. Use a heredoc for multi-line bodies. +- **Read an issue**: `gh issue view --comments`, filtering comments by `jq` and also fetching labels. +- **List issues**: `gh issue list --state open --json number,title,body,labels,comments --jq '[.[] | {number, title, body, labels: [.labels[].name], comments: [.comments[].body]}]'` with appropriate `--label` and `--state` filters. +- **Comment on an issue**: `gh issue comment --body "..."` +- **Apply / remove labels**: `gh issue edit --add-label "..."` / `--remove-label "..."` +- **Close**: `gh issue close --comment "..."` + +Infer the repo from `git remote -v` — `gh` does this automatically when run inside a clone. + +## When a skill says "publish to the issue tracker" + +Create a GitHub issue. + +## When a skill says "fetch the relevant ticket" + +Run `gh issue view --comments`. diff --git a/docs/agents/triage-labels.md b/docs/agents/triage-labels.md new file mode 100644 index 00000000..b716855d --- /dev/null +++ b/docs/agents/triage-labels.md @@ -0,0 +1,15 @@ +# Triage Labels + +The skills speak in terms of five canonical triage roles. This file maps those roles to the actual label strings used in this repo's issue tracker. + +| Label in mattpocock/skills | Label in our tracker | Meaning | +| -------------------------- | -------------------- | ---------------------------------------- | +| `needs-triage` | `needs-triage` | Maintainer needs to evaluate this issue | +| `needs-info` | `needs-info` | Waiting on reporter for more information | +| `ready-for-agent` | `ready-for-agent` | Fully specified, ready for an AFK agent | +| `ready-for-human` | `ready-for-human` | Requires human implementation | +| `wontfix` | `wontfix` | Will not be actioned | + +When a skill mentions a role (e.g. "apply the AFK-ready triage label"), use the corresponding label string from this table. + +Edit the right-hand column to match whatever vocabulary you actually use. diff --git a/landing/components/release-links.generated.ts b/landing/components/release-links.generated.ts index ded3eb90..5d7f1c3d 100644 --- a/landing/components/release-links.generated.ts +++ b/landing/components/release-links.generated.ts @@ -16,7 +16,7 @@ export type PlatformDownload = { assets: PlatformDownloadAsset[]; }; -export const releaseVersion = "0.4.180"; +export const releaseVersion = "0.4.181"; export const platformDownloads: PlatformDownload[] = [ { @@ -29,14 +29,14 @@ export const platformDownloads: PlatformDownload[] = [ architectures: ["arm64"], id: "macos-arm64-dmg", label: "Apple Silicon DMG", - href: "https://pub-a19a3a2c1d2e4fa3a9c120b184024448.r2.dev/AC/AgentCompany-0.4.180-arm64.dmg", + href: "https://pub-a19a3a2c1d2e4fa3a9c120b184024448.r2.dev/AC/AgentCompany-0.4.181-arm64.dmg", note: "arm64", }, { architectures: ["x64"], id: "macos-x64-dmg", label: "Intel Mac DMG", - href: "https://pub-a19a3a2c1d2e4fa3a9c120b184024448.r2.dev/AC/AgentCompany-0.4.180.dmg", + href: "https://pub-a19a3a2c1d2e4fa3a9c120b184024448.r2.dev/AC/AgentCompany-0.4.181.dmg", note: "x64", }, ], @@ -51,7 +51,7 @@ export const platformDownloads: PlatformDownload[] = [ architectures: ["arm64", "x64"], id: "windows-universal-installer", label: "Windows Universal Installer", - href: "https://pub-a19a3a2c1d2e4fa3a9c120b184024448.r2.dev/AC/AgentCompany.Setup.0.4.180.exe", + href: "https://pub-a19a3a2c1d2e4fa3a9c120b184024448.r2.dev/AC/AgentCompany.Setup.0.4.181.exe", note: "arm64 + x64", }, ], @@ -66,42 +66,42 @@ export const platformDownloads: PlatformDownload[] = [ architectures: ["x64"], id: "linux-x64-appimage", label: "Linux x64 AppImage", - href: "https://pub-a19a3a2c1d2e4fa3a9c120b184024448.r2.dev/AC/AgentCompany-0.4.180.AppImage", + href: "https://pub-a19a3a2c1d2e4fa3a9c120b184024448.r2.dev/AC/AgentCompany-0.4.181.AppImage", note: "x64", }, { architectures: ["x64"], id: "linux-x64-deb", label: "Linux x64 DEB", - href: "https://pub-a19a3a2c1d2e4fa3a9c120b184024448.r2.dev/AC/agent-company_0.4.180_amd64.deb", + href: "https://pub-a19a3a2c1d2e4fa3a9c120b184024448.r2.dev/AC/agent-company_0.4.181_amd64.deb", note: "x64", }, { architectures: ["x64"], id: "linux-x64-rpm", label: "Linux x64 RPM", - href: "https://pub-a19a3a2c1d2e4fa3a9c120b184024448.r2.dev/AC/agent-company-0.4.180.x86_64.rpm", + href: "https://pub-a19a3a2c1d2e4fa3a9c120b184024448.r2.dev/AC/agent-company-0.4.181.x86_64.rpm", note: "x64", }, { architectures: ["arm64"], id: "linux-arm64-appimage", label: "Linux ARM64 AppImage", - href: "https://pub-a19a3a2c1d2e4fa3a9c120b184024448.r2.dev/AC/AgentCompany-0.4.180-arm64.AppImage", + href: "https://pub-a19a3a2c1d2e4fa3a9c120b184024448.r2.dev/AC/AgentCompany-0.4.181-arm64.AppImage", note: "arm64", }, { architectures: ["arm64"], id: "linux-arm64-deb", label: "Linux ARM64 DEB", - href: "https://pub-a19a3a2c1d2e4fa3a9c120b184024448.r2.dev/AC/agent-company_0.4.180_arm64.deb", + href: "https://pub-a19a3a2c1d2e4fa3a9c120b184024448.r2.dev/AC/agent-company_0.4.181_arm64.deb", note: "arm64", }, { architectures: ["arm64"], id: "linux-arm64-rpm", label: "Linux ARM64 RPM", - href: "https://pub-a19a3a2c1d2e4fa3a9c120b184024448.r2.dev/AC/agent-company-0.4.180.aarch64.rpm", + href: "https://pub-a19a3a2c1d2e4fa3a9c120b184024448.r2.dev/AC/agent-company-0.4.181.aarch64.rpm", note: "arm64", }, ], diff --git a/package.json b/package.json index 1d837390..930a06d9 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "agent-company", - "version": "0.4.180", + "version": "0.4.182", "private": true, "type": "module", "description": "Cross-platform Electron desktop control plane for local-first AI agent companies.", diff --git a/src/main/agent-runtime/cost-writer.ts b/src/main/agent-runtime/cost-writer.ts index 1e8407c4..45c6f1eb 100644 --- a/src/main/agent-runtime/cost-writer.ts +++ b/src/main/agent-runtime/cost-writer.ts @@ -1,7 +1,7 @@ import { calculateCost, type Usage } from "@mariozechner/pi-ai"; import type { AppDatabase } from "@main/database"; import { getConnectorDefinition } from "@main/connectors"; -import { resolveModel } from "@main/model-resolver"; +import { resolveModel, resolveModelCostFallback } from "@main/model-resolver"; export interface CostWriterDeps { db: AppDatabase; @@ -17,28 +17,38 @@ interface ObserveInput { usage: Usage; } -/** - * Writes a `cost_entries` row for each pi `message_end`. finishRun reads - * SUM(cost_entries.amount_usd) for the run-level aggregate, so this is - * the single source of truth for cost attribution. - */ +// Local OpenAI-compatible servers run on the user's own machine, so the +// LLM call has no third-party billing line. Record a zero cost entry +// without invoking pi-ai's calculator (which would require the model's +// baseUrl from the credential — unnecessary work for a flat-zero cost). +const ZERO_COST_PROVIDERS: ReadonlySet = new Set(["ollama", "lm-studio"]); + export class CostWriter { constructor(private readonly deps: CostWriterDeps) {} observeAssistantMessage(input: ObserveInput): void { - // Fail closed on cost calculation — budget policies read `cost_entries` - // to enforce spending gates; silently emitting an unattributed row - // would let an agent run unbounded. If pricing cannot be resolved, - // throw so the caller logs and the missing cost is visible. const definition = getConnectorDefinition(input.connectorId); if (!definition.provider) { throw new Error(`[cost-writer] connector ${input.connectorId} has no provider mapping`); } + if (ZERO_COST_PROVIDERS.has(definition.provider)) { + this.deps.db.core.runs.recordCostEntry({ + runId: input.runId, + companyId: input.companyId, + agentId: input.agentId, + taskId: input.taskId, + connectorId: input.connectorId, + amountUsd: 0, + unattributed: false, + }); + return; + } const model = resolveModel(definition.provider, input.modelId); if (!model) { throw new Error(`[cost-writer] model ${input.modelId} not found for provider ${definition.provider}`); } - const cost = calculateCost(model, input.usage); + const patchedModel = resolveModelCostFallback(model, definition.provider); + const cost = calculateCost(patchedModel, input.usage); if (!Number.isFinite(cost.total)) { throw new Error( `[cost-writer] cost calculation returned non-finite total for ${input.connectorId}/${input.modelId}`, diff --git a/src/main/agent-runtime/inbox-projection.ts b/src/main/agent-runtime/inbox-projection.ts index 48fbd543..6b1f9e4a 100644 --- a/src/main/agent-runtime/inbox-projection.ts +++ b/src/main/agent-runtime/inbox-projection.ts @@ -83,9 +83,12 @@ export function projectInboxMessageToPi( ): AgentMessage { const senderLabel = resolveSenderLabel(record, directory); const channelLabel = formatChannelLabel(record); + const replyContext = record.channel === "direct" + ? `reply-to: channel=direct, toAgentId=${record.fromAgentId}` + : `reply-to: channel=${record.channel}, channelTargetId=${record.channelTargetId ?? ""}`; const header = channelLabel - ? `[Message from ${senderLabel} · ${channelLabel}]` - : `[Message from ${senderLabel}]`; + ? `[Message from ${senderLabel} · ${channelLabel} | ${replyContext}]` + : `[Message from ${senderLabel} | ${replyContext}]`; const subject = record.subject ? truncate(cleanText(record.subject), MAX_SUBJECT_LENGTH) : ""; const body = truncate(cleanText(record.body), MAX_MESSAGE_BODY_LENGTH); diff --git a/src/main/agent-runtime/runtime-bridge.ts b/src/main/agent-runtime/runtime-bridge.ts index 812a6023..0b4b4e3f 100644 --- a/src/main/agent-runtime/runtime-bridge.ts +++ b/src/main/agent-runtime/runtime-bridge.ts @@ -440,6 +440,7 @@ export class RuntimeBridge { provider: definition.provider, apiKey: resolved.apiKey, extraEnv: resolved.extraEnv ?? {}, + baseUrl: resolved.baseUrl ?? null, priorMessages, }, }; diff --git a/src/main/api-server/routes-agents-standup.ts b/src/main/api-server/routes-agents-standup.ts index f209cb0e..43f8155c 100644 --- a/src/main/api-server/routes-agents-standup.ts +++ b/src/main/api-server/routes-agents-standup.ts @@ -89,6 +89,7 @@ export function registerAgentStandupRoute(server: AgentApiServer): void { id: m.id, from: senderNames.get(m.fromAgentId) ?? m.fromAgentId, fromAgentId: m.fromAgentId, + toAgentId: m.toAgentId, channel: m.channel, channelTargetId: m.channelTargetId, subject: m.subject, diff --git a/src/main/automation/automation-approval-handler.ts b/src/main/automation/automation-approval-handler.ts index 0506ea7f..b779ec69 100644 --- a/src/main/automation/automation-approval-handler.ts +++ b/src/main/automation/automation-approval-handler.ts @@ -83,6 +83,13 @@ export function handleApprovalResolved( if (deps.isQuitting()) return; const { db, logger, publishDomainChanged, notify, wakeAgentIfPossible, queueBrowserActionExecution } = deps; const { dispatchAutomation, handleTaskStatusChange, runWorkflowPipeline, trackTimer } = callbacks; + const explicitlyWokenAgentIds = new Set(); + + function wakeAgent(agentId: string | null | undefined, trigger: string): void { + if (!agentId) return; + wakeAgentIfPossible(agentId, companyId, trigger); + explicitlyWokenAgentIds.add(agentId); + } const snapshot = db.projection.snapshot.listSnapshot(); const approval = snapshot.approvals.find((entry) => entry.id === approvalId); @@ -134,11 +141,11 @@ export function handleApprovalResolved( companyId, agentId: agent.id, }); - wakeAgentIfPossible(agent.id, companyId, "assignment"); + wakeAgent(agent.id, "assignment"); const ceo = db.core.agents.getCeoForCompany(companyId); if (ceo && ceo.id !== agent.id) { - wakeAgentIfPossible(ceo.id, companyId, "assignment"); + wakeAgent(ceo.id, "assignment"); } publishDomainChanged(); @@ -186,20 +193,15 @@ export function handleApprovalResolved( db.core.tasks.advanceTaskStatus(task.id, "done", "Board approved deliverable."); handleTaskStatusChange(task.id, companyId, task.status, "done"); db.core.activity.addActivity({ companyId, actor: "board", action: "deliverable.approved", entityType: "task", entityId: task.id, detail: approval.decisionNote || "Deliverable approved by board." }); - } else if (decision === "rejected") { + } else if (decision === "rejected" || decision === "revision_requested") { + const previousStatus = task.status; + const statusMessage = decision === "rejected" + ? `Board rejected deliverable: ${approval.decisionNote || "No reason given."}` + : `Board requested revisions: ${approval.decisionNote || "No reason given."}`; recordReviewFeedbackKnowledge(db, approval, decision); - db.core.tasks.advanceTaskStatus(task.id, "in_progress", `Board rejected deliverable: ${approval.decisionNote || "No reason given."}`); - handleTaskStatusChange(task.id, companyId, task.status, "in_progress"); - if (task.assigneeAgentId) { - wakeAgentIfPossible(task.assigneeAgentId, companyId, "revision_needed"); - } - } else if (decision === "revision_requested") { - db.core.tasks.advanceTaskStatus(task.id, "in_progress", `Board requested revisions: ${approval.decisionNote || "No reason given."}`); - handleTaskStatusChange(task.id, companyId, task.status, "in_progress"); - recordReviewFeedbackKnowledge(db, approval, decision); - if (task.assigneeAgentId) { - wakeAgentIfPossible(task.assigneeAgentId, companyId, "revision_needed"); - } + db.core.tasks.advanceTaskStatus(task.id, "todo", statusMessage); + handleTaskStatusChange(task.id, companyId, previousStatus, "todo"); + wakeAgent(task.assigneeAgentId, "revision_needed"); } } else if (approval.requestedByAgentId) { const agent = db.core.entity.getAgent(approval.requestedByAgentId); @@ -208,7 +210,7 @@ export function handleApprovalResolved( logger.info(`[deliverable-review] Resumed heartbeat for ${agent.name} after user ${decision}.`); } if (decision !== "approved") { - wakeAgentIfPossible(approval.requestedByAgentId, companyId, "revision_needed"); + wakeAgent(approval.requestedByAgentId, "revision_needed"); } } @@ -366,6 +368,8 @@ export function handleApprovalResolved( } if (approval.requestedByAgentId) { - wakeAgentIfPossible(approval.requestedByAgentId, companyId, "approval_resolved"); + if (!explicitlyWokenAgentIds.has(approval.requestedByAgentId)) { + wakeAgentIfPossible(approval.requestedByAgentId, companyId, "approval_resolved"); + } } } diff --git a/src/main/budget-enforcement-service.ts b/src/main/budget-enforcement-service.ts index df934143..6c4bac83 100644 --- a/src/main/budget-enforcement-service.ts +++ b/src/main/budget-enforcement-service.ts @@ -157,13 +157,11 @@ export function createBudgetEnforcementService({ const spent = db.core.budget.getSpentInWindow(policy); let resolved = false; - // If spend is below hard-stop threshold, resolve all open incidents if (spent < policy.amount) { const count = db.core.budget.resolveIncidentsForPolicy(policyId, "auto_resolved_limit_increase"); if (count > 0) resolved = true; } - // If spend is below warn threshold, resolve warn incidents specifically const warnAmount = Math.floor((policy.amount * policy.warnPercent) / 100); if (spent < warnAmount) { const count = db.core.budget.resolveWarnIncidentsForPolicy(policyId, "auto_resolved_limit_increase"); @@ -171,11 +169,26 @@ export function createBudgetEnforcementService({ } if (resolved) { + unpauseBudgetBlockedAgents(companyId); logger.info(`Budget auto-resolve: policy ${policyId} incidents resolved (spent=${spent}, limit=${policy.amount})`); publishDomainChanged(); } } + function unpauseBudgetBlockedAgents(companyId: string): void { + const paused = db.queryAll<{ id: string; budgetMonthlyUsd: number; spentMonthlyUsd: number }>( + `SELECT id, budget_monthly_usd as budgetMonthlyUsd, spent_monthly_usd as spentMonthlyUsd + FROM agents WHERE company_id = ? AND status = 'paused' AND budget_monthly_usd > 0`, + companyId, + ); + for (const agent of paused) { + if (agent.budgetMonthlyUsd > agent.spentMonthlyUsd) { + db.queryAll("UPDATE agents SET status = 'idle', updated_at = ? WHERE id = ?", new Date().toISOString(), agent.id); + logger.info(`[budget] Auto-unpaused agent ${agent.id}: budget $${agent.budgetMonthlyUsd} > spent $${agent.spentMonthlyUsd}`); + } + } + } + /** * Closes incidents whose calendar month window has ended. * Per D-06: window rollover cleanup. diff --git a/src/main/connector-definitions.ts b/src/main/connector-definitions.ts index f906672f..cdd29e85 100644 --- a/src/main/connector-definitions.ts +++ b/src/main/connector-definitions.ts @@ -25,7 +25,7 @@ import type { KnownProvider } from "@mariozechner/pi-ai"; import type { ConnectorId } from "@shared/types"; -import { OAUTH_CONNECTORS, API_KEY_CONNECTORS, CLOUD_CONNECTORS } from "./connector-definitions/index"; +import { OAUTH_CONNECTORS, API_KEY_CONNECTORS, CLOUD_CONNECTORS, LOCAL_CONNECTORS } from "./connector-definitions/index"; /** * OAuth provider id recognised by the AgentCompany credential layer. These @@ -61,14 +61,21 @@ export interface ConnectorAuthGcpAdc { type: "gcp_adc"; } +export interface ConnectorAuthLocal { + type: "local"; + /** Pre-filled base URL the onboarding card shows on first save. */ + defaultBaseUrl: string; +} + export type ConnectorAuth = | ConnectorAuthOAuth | ConnectorAuthApiKey | ConnectorAuthAws | ConnectorAuthAzure - | ConnectorAuthGcpAdc; + | ConnectorAuthGcpAdc + | ConnectorAuthLocal; -export type LocalModelResolverProvider = "fireworks" | "fireworks-firepass"; +export type LocalModelResolverProvider = "fireworks" | "fireworks-firepass" | "ollama" | "lm-studio"; export type ConnectorProvider = KnownProvider | LocalModelResolverProvider; export interface ConnectorDefinition { @@ -94,4 +101,5 @@ export const CONNECTORS: ConnectorDefinition[] = [ ...OAUTH_CONNECTORS, ...API_KEY_CONNECTORS, ...CLOUD_CONNECTORS, + ...LOCAL_CONNECTORS, ]; diff --git a/src/main/connector-definitions/apikey.ts b/src/main/connector-definitions/apikey.ts index 2d6620c3..db6a2aa6 100644 --- a/src/main/connector-definitions/apikey.ts +++ b/src/main/connector-definitions/apikey.ts @@ -168,10 +168,10 @@ export const API_KEY_CONNECTORS: ConnectorDefinition[] = [ provider: "fireworks-firepass", label: "Fireworks Fire Pass", description: - "Fireworks Fire Pass — $7/week subscription for unlimited Kimi K2.5 Turbo. Uses the same FIREWORKS_API_KEY; the Fire Pass entitlement is attached to the account.", - defaultModel: "accounts/fireworks/routers/kimi-k2p5-turbo", + "Fireworks Fire Pass — $7/week subscription for unlimited Kimi K2.6 Turbo. Uses the same FIREWORKS_API_KEY; the Fire Pass entitlement is attached to the account.", + defaultModel: "accounts/fireworks/routers/kimi-k2p6-turbo", auth: { type: "api_key", envKey: "FIREWORKS_API_KEY" }, notes: - "Fire Pass routes through the dedicated router id accounts/fireworks/routers/kimi-k2p5-turbo on both the OpenAI- and Anthropic-compat endpoints. See https://docs.fireworks.ai/firepass.", + "Fire Pass routes through the dedicated router id accounts/fireworks/routers/kimi-k2p6-turbo on both the OpenAI- and Anthropic-compat endpoints. See https://docs.fireworks.ai/firepass.", }, ]; diff --git a/src/main/connector-definitions/index.ts b/src/main/connector-definitions/index.ts index 51e9f64e..380227e0 100644 --- a/src/main/connector-definitions/index.ts +++ b/src/main/connector-definitions/index.ts @@ -9,3 +9,4 @@ export { OAUTH_CONNECTORS } from "./oauth"; export { API_KEY_CONNECTORS } from "./apikey"; export { CLOUD_CONNECTORS } from "./cloud"; +export { LOCAL_CONNECTORS } from "./local"; diff --git a/src/main/connector-definitions/local.ts b/src/main/connector-definitions/local.ts new file mode 100644 index 00000000..c538647a --- /dev/null +++ b/src/main/connector-definitions/local.ts @@ -0,0 +1,41 @@ +/** + * Local pi-mono connectors. + * + * Ollama and LM Studio both expose an OpenAI-compatible + * `/v1/chat/completions` endpoint, so they are routed through pi-ai's + * `openai-completions` API path with a per-installation `baseUrl` that + * the user enters during onboarding. The provider ids are deliberately + * absent from `@mariozechner/pi-ai`'s `KnownProvider` set — model + * dispatch resolves them locally via `src/main/model-resolver.ts` the + * same way `fireworks` and `fireworks-firepass` do. + * + * No environment variable name accompanies the auth descriptor: a local + * server's address is a user choice, never read from `process.env`. + */ + +import type { ConnectorDefinition } from "../connector-definitions"; + +export const LOCAL_CONNECTORS: ConnectorDefinition[] = [ + { + id: "ollama", + provider: "ollama", + label: "Ollama", + description: + "Local Ollama server (https://ollama.com) — runs open-weights models on your own machine through its OpenAI-compatible endpoint.", + defaultModel: "llama3.2", + auth: { type: "local", defaultBaseUrl: "http://localhost:11434/v1" }, + notes: + "Ollama exposes /v1/chat/completions and /v1/models. The default port is 11434; override if you launched `ollama serve` on a different address.", + }, + { + id: "lm-studio", + provider: "lm-studio", + label: "LM Studio", + description: + "Local LM Studio server (https://lmstudio.ai) — host any loaded model through its OpenAI-compatible endpoint.", + defaultModel: "local-model", + auth: { type: "local", defaultBaseUrl: "http://localhost:1234/v1" }, + notes: + "Enable LM Studio's local server from the Developer tab; the default port is 1234. Model ids match whatever you have loaded.", + }, +]; diff --git a/src/main/connector-health-service.ts b/src/main/connector-health-service.ts index 26228fcf..bfe4a085 100644 --- a/src/main/connector-health-service.ts +++ b/src/main/connector-health-service.ts @@ -109,6 +109,15 @@ export function createConnectorHealthService({ } return { status: "ready", authState: "ok", lastError: null }; } + case "local": { + if (stored?.kind !== "local") { + return { status: "auth_required", authState: "required", lastError: "Local server URL not configured." }; + } + if (!stored.value.baseUrl) { + return { status: "error", authState: "error", lastError: "Local server URL missing." }; + } + return { status: "ready", authState: "ok", lastError: null }; + } } } diff --git a/src/main/connectors.ts b/src/main/connectors.ts index 20a1bc4e..34858856 100644 --- a/src/main/connectors.ts +++ b/src/main/connectors.ts @@ -21,6 +21,8 @@ const RESERVED_ENV_BINDING_KEYS = new Set([ const LOCAL_MODEL_RESOLVER_PROVIDERS: ReadonlySet = new Set([ "fireworks", "fireworks-firepass", + "ollama", + "lm-studio", ]); const PI_KNOWN_PROVIDER_IDS: ReadonlySet = new Set(getProviders()); @@ -110,6 +112,11 @@ function renderConfigurationDoc(definition: ConnectorDefinition): string { definition.label, "Uses Google Cloud Application Default Credentials. Run `gcloud auth application-default login` or set GOOGLE_APPLICATION_CREDENTIALS.", ); + case "local": + return buildCloudDoc( + definition.label, + `OpenAI-compatible local server. Default base URL: ${definition.auth.defaultBaseUrl}. Configure under Onboarding → Providers (Advanced) and pick a model id that your server actually serves.`, + ); } } diff --git a/src/main/credentials/credential-resolver.ts b/src/main/credentials/credential-resolver.ts index 2c719c64..5fa617da 100644 --- a/src/main/credentials/credential-resolver.ts +++ b/src/main/credentials/credential-resolver.ts @@ -34,6 +34,8 @@ export class CredentialError extends UpstreamError { export interface ResolvedCredential { apiKey: string; extraEnv?: Record; + /** Base URL for OpenAI-compatible local servers (Ollama, LM Studio). */ + baseUrl?: string; } export class CredentialResolver { @@ -144,6 +146,16 @@ export class CredentialResolver { if (cfg.credentialsPath) env.GOOGLE_APPLICATION_CREDENTIALS = cfg.credentialsPath; return { apiKey: "", extraEnv: env }; } + case "local": { + if (stored?.kind !== "local") { + throw new CredentialError("CREDENTIAL_MISSING", connectorId); + } + // Ollama and LM Studio ignore the bearer value but the OpenAI SDK + // refuses an empty string, so we hand it the connector id as a + // stable non-empty placeholder. The actual transport is keyed off + // `baseUrl`, surfaced to the worker through the stream model. + return { apiKey: connectorId, baseUrl: stored.value.baseUrl }; + } default: { const exhaustive: never = auth; throw new CredentialError("CREDENTIAL_INVALID", connectorId, `Unsupported auth descriptor: ${JSON.stringify(exhaustive)}`); diff --git a/src/main/credentials/credential-store.ts b/src/main/credentials/credential-store.ts index 92f0f0b2..9a88c1ea 100644 --- a/src/main/credentials/credential-store.ts +++ b/src/main/credentials/credential-store.ts @@ -39,6 +39,8 @@ export function deriveDisplayHint(payload: StoredCredentialPayload): string { : "no-expiry"; return `oauth:expires=${expiresLabel}`; } + case "local": + return `local:${payload.value.baseUrl}`; } } diff --git a/src/main/credentials/types.ts b/src/main/credentials/types.ts index aaa6e7b9..a247bab3 100644 --- a/src/main/credentials/types.ts +++ b/src/main/credentials/types.ts @@ -3,6 +3,7 @@ import type { AwsCredential, AzureCredential, GcpAdcCredential, + LocalCredential, } from "@shared/contracts/schemas-credentials"; /** @@ -25,6 +26,7 @@ export type StoredCredentialPayload = | { kind: "api_key"; value: ApiKeyValue } | { kind: "azure"; value: AzureCredential } | { kind: "aws"; value: AwsCredential } - | { kind: "gcp_adc"; value: GcpAdcCredential }; + | { kind: "gcp_adc"; value: GcpAdcCredential } + | { kind: "local"; value: LocalCredential }; export type CredentialKind = StoredCredentialPayload["kind"]; diff --git a/src/main/database/approvals.ts b/src/main/database/approvals.ts index 947e9341..25da7786 100644 --- a/src/main/database/approvals.ts +++ b/src/main/database/approvals.ts @@ -415,9 +415,10 @@ export function buildApprovalMethods( .prepare( `SELECT 1 as hit FROM approvals a JOIN tasks t ON a.related_task_id = t.id + JOIN approval_policies p ON p.approval_type = a.type WHERE a.company_id = ? AND a.state = 'pending' AND t.assignee_agent_id = ? - AND a.type IN ('plan_review','scope_review','draft_review','strategy_review','deliverable_review','exception_review') + AND p.blocks_execution = 1 LIMIT 1`, ) .get(companyId, agentId) as { hit: number } | undefined; diff --git a/src/main/database/credentials.ts b/src/main/database/credentials.ts index 6b4f52cf..1c8b8692 100644 --- a/src/main/database/credentials.ts +++ b/src/main/database/credentials.ts @@ -2,7 +2,7 @@ import type { DatabaseSync } from "node:sqlite"; import type { ReadAll, Transaction } from "./helpers"; import { nowIso } from "./helpers"; -export type CredentialKind = "oauth" | "api_key" | "aws" | "azure" | "gcp_adc"; +export type CredentialKind = "oauth" | "api_key" | "aws" | "azure" | "gcp_adc" | "local"; export type CredentialStatus = "active" | "expired" | "revoked" | "error"; export type CredentialBackend = "safe_storage" | "vault"; diff --git a/src/main/database/entity-crud.ts b/src/main/database/entity-crud.ts index 7451831d..9320a93b 100644 --- a/src/main/database/entity-crud.ts +++ b/src/main/database/entity-crud.ts @@ -1,7 +1,7 @@ import type { DatabaseSync } from "node:sqlite"; import type { AgentRecord, ConnectorId, TaskRecord } from "@shared/types"; import type { ReadAll, Transaction } from "./helpers"; -import { ensureId, normalizeOptionalId, nowIso } from "./helpers"; +import { activeRunIdSql, ensureId, normalizeOptionalId, nowIso } from "./helpers"; /** * Goal row shape returned by listGoals / getGoalForUpdate. Mirrors the @@ -53,7 +53,7 @@ export function buildEntityCrudMethods( getTask(id: string): TaskRecord { const row = getDb() .prepare( - "select id, company_id as companyId, project_id as projectId, goal_id as goalId, parent_id as parentId, title, description, assignee_agent_id as assigneeAgentId, workspace_id as workspaceId, status, priority, (select l.run_id from task_leases l where l.task_id = tasks.id) as activeRunId, cost_usd as costUsd, task_type as taskType, requires_user_review as requiresUserReview, requires_filesystem as requiresFilesystem, started_at as startedAt, completed_at as completedAt, cancelled_at as cancelledAt, created_at as createdAt, updated_at as updatedAt from tasks where id = ?", + `select id, company_id as companyId, project_id as projectId, goal_id as goalId, parent_id as parentId, title, description, assignee_agent_id as assigneeAgentId, workspace_id as workspaceId, status, priority, ${activeRunIdSql("tasks")} as activeRunId, cost_usd as costUsd, task_type as taskType, requires_user_review as requiresUserReview, requires_filesystem as requiresFilesystem, started_at as startedAt, completed_at as completedAt, cancelled_at as cancelledAt, created_at as createdAt, updated_at as updatedAt from tasks where id = ?`, ) .get(id) as TaskRecord | undefined; if (!row) throw new Error(`Task not found: ${id}`); diff --git a/src/main/database/helpers.ts b/src/main/database/helpers.ts index 3a4ddfc4..80f4c21c 100644 --- a/src/main/database/helpers.ts +++ b/src/main/database/helpers.ts @@ -42,3 +42,12 @@ export function taskPriorityOrderSql(columnName: string): string { else 0 end`; } + +export function activeRunIdSql(taskTableAlias: string): string { + return `(select l.run_id + from task_leases l + join runs r on r.id = l.run_id + where l.task_id = ${taskTableAlias}.id + and r.status in ('queued', 'running') + limit 1)`; +} diff --git a/src/main/database/messages.ts b/src/main/database/messages.ts index f17606d7..8de63874 100644 --- a/src/main/database/messages.ts +++ b/src/main/database/messages.ts @@ -70,10 +70,13 @@ export function buildMessageMethods( const body = normalizeMarkdownText(input.body); const id = randomUUID(); const createdAt = nowIso(); - // Single transaction: the row and its mention fan-out must land - // atomically, otherwise a crash between the two writes leaves a - // message visible to its recipient but silently invisible to the - // mentioned agents — which is exactly the old bug we're fixing. + // Single transaction: the row, its self-read receipt, and its + // mention fan-out must land atomically. The self-read receipt + // makes "author has read their own message" an invariant in the + // join table — without it, every message a sender posts inflates + // their own sidebar / inbox / channel unread counts until they + // happen to re-open the channel, because `readAt` is derived from + // `agent_message_reads` per viewer. deps.transaction(() => { getDb() .prepare( @@ -94,6 +97,11 @@ export function buildMessageMethods( input.attachmentsJson ?? "[]", createdAt, ); + getDb() + .prepare( + "insert into agent_message_reads (message_id, agent_id, read_at) values (?, ?, ?)", + ) + .run(id, input.fromAgentId, createdAt); if (input.mentionedAgentIds && input.mentionedAgentIds.length > 0) { insertMessageMentions(id, input.companyId, input.fromAgentId, input.mentionedAgentIds, createdAt); } diff --git a/src/main/database/runs.ts b/src/main/database/runs.ts index 45e444b8..d84f362c 100644 --- a/src/main/database/runs.ts +++ b/src/main/database/runs.ts @@ -166,20 +166,20 @@ export function buildRunMethods( model: string | null; unattributedCost: boolean; usageStatus?: RunUsageStatus; + recoverTaskForRetry?: boolean; }) { const run = getDb().prepare("select task_id as taskId, agent_id as agentId, company_id as companyId, connector_id as connectorId, started_at as startedAt from runs where id = ?").get(input.runId) as | { taskId: string | null; agentId: string; companyId: string; connectorId: ConnectorId; startedAt: string | null } | undefined; if (!run) return { budgetHardStopped: false }; - // A run that never transitioned to 'running' (startedAt === null) and - // is now terminal with status 'failed' / 'cancelled' must roll the - // task back to 'todo'. Otherwise the task is stuck in 'in_progress' - // with no owner — checkoutTask refuses (WHERE status IN todo/backlog), - // and the scheduler loops forever logging "lost checkout race". The - // common trigger is workspace-lock contention: `dispatchRunToWorker` - // bails before work starts and calls finishRun('failed'). + // Recoverable terminal paths must roll the task back to 'todo'. + // Otherwise the task is stuck in 'in_progress' with no owner: + // checkoutTask refuses (WHERE status IN todo/backlog), and the + // scheduler loops forever logging "lost checkout race". const runNeverStarted = run.startedAt === null && (input.status === "failed" || input.status === "cancelled"); + const shouldReturnTaskToTodo = + run.taskId !== null && (runNeverStarted || input.recoverTaskForRetry === true); const ts = nowIso(); const usageStatus = input.usageStatus ?? deriveLegacyRunUsageStatus({ runStatus: input.status, @@ -220,7 +220,7 @@ export function buildRunMethods( getDb() .prepare("delete from task_leases where task_id = ? and run_id = ?") .run(run.taskId, input.runId); - if (runNeverStarted) { + if (shouldReturnTaskToTodo) { // Roll the task back to `todo` when no *other* live run claims // it. Invariant: `task.status = 'in_progress'` ⇒ a queued or // running run owns it through `task_leases`. diff --git a/src/main/database/snapshot.ts b/src/main/database/snapshot.ts index 7d2d3810..6dccf599 100644 --- a/src/main/database/snapshot.ts +++ b/src/main/database/snapshot.ts @@ -15,7 +15,7 @@ import type { TaskRecord, WorkspaceRecord, } from "@shared/types"; -import { BOARD_READER_ID, type ReadAll, type SqliteRow, toBoolean } from "./helpers"; +import { activeRunIdSql, BOARD_READER_ID, type ReadAll, type SqliteRow, toBoolean } from "./helpers"; export function buildSnapshotMethods( getDb: () => DatabaseSync, @@ -83,7 +83,7 @@ export function buildSnapshotMethods( ...companyArgs, ), tasks: readAll( - `select id, company_id as companyId, project_id as projectId, goal_id as goalId, parent_id as parentId, title, description, assignee_agent_id as assigneeAgentId, workspace_id as workspaceId, status, priority, (select l.run_id from task_leases l where l.task_id = tasks.id) as activeRunId, cost_usd as costUsd, task_type as taskType, requires_user_review as requiresUserReview, requires_filesystem as requiresFilesystem, started_at as startedAt, completed_at as completedAt, cancelled_at as cancelledAt, created_at as createdAt, updated_at as updatedAt from tasks${companyWhere} order by created_at desc limit 2000`, + `select id, company_id as companyId, project_id as projectId, goal_id as goalId, parent_id as parentId, title, description, assignee_agent_id as assigneeAgentId, workspace_id as workspaceId, status, priority, ${activeRunIdSql("tasks")} as activeRunId, cost_usd as costUsd, task_type as taskType, requires_user_review as requiresUserReview, requires_filesystem as requiresFilesystem, started_at as startedAt, completed_at as completedAt, cancelled_at as cancelledAt, created_at as createdAt, updated_at as updatedAt from tasks${companyWhere} order by created_at desc limit 2000`, ...companyArgs, ).map((record) => ({ ...record, requiresUserReview: toBoolean(record.requiresUserReview), requiresFilesystem: toBoolean(record.requiresFilesystem), startedAt: record.startedAt ?? null, completedAt: record.completedAt ?? null, cancelledAt: record.cancelledAt ?? null })) as TaskRecord[], approvals: readAll( diff --git a/src/main/database/task-queries.ts b/src/main/database/task-queries.ts index 52f6aeab..a1a6838c 100644 --- a/src/main/database/task-queries.ts +++ b/src/main/database/task-queries.ts @@ -1,6 +1,6 @@ import type { DatabaseSync } from "node:sqlite"; import type { TaskRecord } from "@shared/types"; -import { type ReadAll, type Transaction, taskPriorityOrderSql } from "./helpers"; +import { activeRunIdSql, type ReadAll, type Transaction, taskPriorityOrderSql } from "./helpers"; import type { TaskListByStatusesRow } from "./tasks-crud"; /** @@ -25,7 +25,7 @@ export function buildTaskQueryMethods( ): TaskRecord[] { const placeholders = statuses.map(() => "?").join(","); return readAll( - `select id, company_id as companyId, project_id as projectId, goal_id as goalId, parent_id as parentId, title, description, assignee_agent_id as assigneeAgentId, workspace_id as workspaceId, status, priority, (select l.run_id from task_leases l where l.task_id = tasks.id) as activeRunId, cost_usd as costUsd, task_type as taskType, created_at as createdAt, updated_at as updatedAt + `select id, company_id as companyId, project_id as projectId, goal_id as goalId, parent_id as parentId, title, description, assignee_agent_id as assigneeAgentId, workspace_id as workspaceId, status, priority, ${activeRunIdSql("tasks")} as activeRunId, cost_usd as costUsd, task_type as taskType, created_at as createdAt, updated_at as updatedAt from tasks where company_id = ? and assignee_agent_id = ? and status in (${placeholders}) order by @@ -91,7 +91,7 @@ export function buildTaskQueryMethods( ): TaskListByStatusesRow[] { const placeholders = statuses.map(() => "?").join(", "); return readAll( - `select id, company_id as companyId, project_id as projectId, goal_id as goalId, parent_id as parentId, title, description, assignee_agent_id as assigneeAgentId, workspace_id as workspaceId, status, priority, (select l.run_id from task_leases l where l.task_id = tasks.id) as activeRunId, cost_usd as costUsd, requires_user_review as requiresUserReview, started_at as startedAt, completed_at as completedAt, cancelled_at as cancelledAt, created_at as createdAt, updated_at as updatedAt from tasks where company_id = ? and status in (${placeholders}) order by created_at desc`, + `select id, company_id as companyId, project_id as projectId, goal_id as goalId, parent_id as parentId, title, description, assignee_agent_id as assigneeAgentId, workspace_id as workspaceId, status, priority, ${activeRunIdSql("tasks")} as activeRunId, cost_usd as costUsd, requires_user_review as requiresUserReview, started_at as startedAt, completed_at as completedAt, cancelled_at as cancelledAt, created_at as createdAt, updated_at as updatedAt from tasks where company_id = ? and status in (${placeholders}) order by created_at desc`, companyId, ...statuses, ).map((t) => ({ diff --git a/src/main/database/tasks-status.ts b/src/main/database/tasks-status.ts index 468c8ad8..66855d21 100644 --- a/src/main/database/tasks-status.ts +++ b/src/main/database/tasks-status.ts @@ -3,6 +3,7 @@ import type { TaskRecord } from "@shared/types"; import { type ReadAll, type Transaction, + activeRunIdSql, nowIso, taskPriorityOrderSql, } from "./helpers"; @@ -82,7 +83,7 @@ export function buildTaskStatusMethods( t.parent_id as parentId, t.title, t.description, t.assignee_agent_id as assigneeAgentId, t.workspace_id as workspaceId, t.status, t.priority, - (select l.run_id from task_leases l where l.task_id = t.id) as activeRunId, + ${activeRunIdSql("t")} as activeRunId, t.cost_usd as costUsd, t.created_at as createdAt, t.updated_at as updatedAt from tasks t diff --git a/src/main/ipc-core-agent-graph-handlers.ts b/src/main/ipc-core-agent-graph-handlers.ts index d900283c..cd811822 100644 --- a/src/main/ipc-core-agent-graph-handlers.ts +++ b/src/main/ipc-core-agent-graph-handlers.ts @@ -99,6 +99,14 @@ export function registerCoreAgentGraphHandlers({ if (!modelSelection.ok) { return fail("CONNECTOR_MODEL_NOT_ALLOWED", modelSelection.error); } + let resolvedStatus = parsed.status; + if (resolvedStatus === "paused" && existingAgent.status === "paused" && parsed.budgetMonthlyUsd > 0) { + const spent = existingAgent.spentMonthlyUsd ?? 0; + if (parsed.budgetMonthlyUsd > spent) { + resolvedStatus = "idle"; + } + } + let id: string; try { id = db.core.agents.saveAgent({ @@ -108,7 +116,7 @@ export function registerCoreAgentGraphHandlers({ role: parsed.role, title: parsed.title, department: parsed.department ?? null, - status: parsed.status, + status: resolvedStatus, reportsTo: parsed.reportsTo ?? null, connectorId: parsed.connectorId, workspaceId: parsed.workspaceId ?? null, @@ -116,9 +124,6 @@ export function registerCoreAgentGraphHandlers({ metadataJson: parsed.metadataJson, capabilities: parsed.capabilities, budgetMonthlyUsd: parsed.budgetMonthlyUsd, - // User edited this agent through the form and picked a specific - // (connector, model). Respect their intent: pin the strategy to - // `explicit` so the dispatcher never overrides their choice. modelStrategy: "explicit", }); } catch (error) { diff --git a/src/main/ipc-credentials-handlers.ts b/src/main/ipc-credentials-handlers.ts index 0ec38799..051e241e 100644 --- a/src/main/ipc-credentials-handlers.ts +++ b/src/main/ipc-credentials-handlers.ts @@ -177,7 +177,7 @@ export function registerCredentialIpcHandlers(deps: CredentialHandlerDeps): () = } const provider = definition.provider; const defaultModel = definition.defaultModel; - const model = resolveModel(provider, defaultModel); + const model = resolveModel(provider, defaultModel, { baseUrl: resolved.baseUrl }); if (!model) return ok({ ok: false, message: `Unknown model ${defaultModel} for provider ${provider}` }); const extraHeaders = resolveExtraStreamHeaders(provider, resolved.apiKey); const options: SimpleStreamOptions = { diff --git a/src/main/main-runtime-support.ts b/src/main/main-runtime-support.ts index fabda301..f1862dab 100644 --- a/src/main/main-runtime-support.ts +++ b/src/main/main-runtime-support.ts @@ -184,8 +184,18 @@ export function createMainRuntimeSupport(state: MainRuntimeState) { // user has selected a company; the onboarding path uses this // catalog directly because no company exists yet. const { getConnectorDefinition } = await import("./connectors"); - const { resolveModels } = await import("./model-resolver"); + const { resolveModels, fetchLocalConnectorModels } = await import("./model-resolver"); const definition = getConnectorDefinition(connectorId as ConnectorId); + if (definition.auth.type === "local") { + // Local servers' catalogs depend on the user's installed/loaded + // models. Read the saved baseUrl and ask the server directly. + const piRuntime = requireServiceGraph().getPiRuntime(); + const credentials = piRuntime?.credentialStore.read(connectorId); + const baseUrl = credentials?.kind === "local" ? credentials.value.baseUrl : null; + if (!baseUrl) return []; + const models = await fetchLocalConnectorModels(baseUrl); + return models.map((m) => ({ id: m.id, label: m.name })); + } return resolveModels(definition.provider).map((m) => ({ id: m.id, label: m.name })); }, logger: log, diff --git a/src/main/model-capability-seed.ts b/src/main/model-capability-seed.ts index beb2b11f..f6648309 100644 --- a/src/main/model-capability-seed.ts +++ b/src/main/model-capability-seed.ts @@ -103,10 +103,10 @@ export const MODEL_CAPABILITY_SEED: Array> = tier: "standard", costPerMtokOut: 3, }, - // Kimi K2.5 Turbo via Fire Pass router — newer, faster, similar quality. + // Kimi K2.6 Turbo via Fire Pass router — newer, faster, similar quality. { provider: "fireworks-firepass", - modelId: "accounts/fireworks/routers/kimi-k2p5-turbo", + modelId: "accounts/fireworks/routers/kimi-k2p6-turbo", reasoning: 4, coding: 4, writing: 5, diff --git a/src/main/model-resolver.ts b/src/main/model-resolver.ts index 7d1c1464..5895330d 100644 --- a/src/main/model-resolver.ts +++ b/src/main/model-resolver.ts @@ -34,18 +34,47 @@ const FIREWORKS_DEFAULT_MAX_TOKENS = 8192; // `accounts/fireworks/models/...` id — the subscription entitlement // only attaches to the router route, and switching to the model path // makes Fireworks bill per-token and reject Fire-Pass-only callers. -export const FIREWORKS_FIREPASS_MODEL_ID = "accounts/fireworks/routers/kimi-k2p5-turbo"; -export const FIREWORKS_FIREPASS_MODEL_NAME = "Kimi K2.5 Turbo (Fire Pass)"; +export const FIREWORKS_FIREPASS_MODEL_ID = "accounts/fireworks/routers/kimi-k2p6-turbo"; +export const FIREWORKS_FIREPASS_MODEL_NAME = "Kimi K2.6 Turbo (Fire Pass)"; -const FIREWORKS_CURATED_MODELS: Array<{ id: string; name: string; contextWindow?: number }> = [ - { id: "accounts/fireworks/models/kimi-k2-instruct", name: "Kimi K2 Instruct", contextWindow: 131_072 }, - { id: "accounts/fireworks/models/qwen3-coder-480b-a35b-instruct", name: "Qwen3 Coder 480B A35B Instruct", contextWindow: 262_144 }, - { id: "accounts/fireworks/models/deepseek-v3p1", name: "DeepSeek V3.1", contextWindow: 163_840 }, - { id: "accounts/fireworks/models/glm-4p5", name: "GLM 4.5", contextWindow: 131_072 }, - { id: "accounts/fireworks/models/llama-v3p3-70b-instruct", name: "Llama 3.3 70B Instruct", contextWindow: 131_072 }, +const FIREWORKS_CURATED_MODELS: Array<{ + id: string; + name: string; + contextWindow?: number; + cost: { input: number; output: number }; +}> = [ + { id: "accounts/fireworks/models/kimi-k2-instruct", name: "Kimi K2 Instruct", contextWindow: 131_072, cost: { input: 0.50, output: 0.50 } }, + { id: "accounts/fireworks/models/qwen3-coder-480b-a35b-instruct", name: "Qwen3 Coder 480B A35B Instruct", contextWindow: 262_144, cost: { input: 0.50, output: 0.50 } }, + { id: "accounts/fireworks/models/deepseek-v3p1", name: "DeepSeek V3.1", contextWindow: 163_840, cost: { input: 0.20, output: 0.20 } }, + { id: "accounts/fireworks/models/glm-4p5", name: "GLM 4.5", contextWindow: 131_072, cost: { input: 0.30, output: 0.30 } }, + { id: "accounts/fireworks/models/llama-v3p3-70b-instruct", name: "Llama 3.3 70B Instruct", contextWindow: 131_072, cost: { input: 0.20, output: 0.20 } }, ]; -function buildFireworksModel(modelId: string, name?: string, contextWindow?: number): Model<"openai-completions"> { +const FIREWORKS_FIREPASS_COST = { input: 0, output: 0 }; +const FIREWORKS_FALLBACK_COST = { input: 0.50, output: 0.50 }; + +// Local OpenAI-compatible servers (Ollama, LM Studio) bill the user's own +// hardware, so the surfaced cost is zero. Context window varies wildly +// across self-hosted models; the configured window is conservative +// enough to keep pi's compactor honest without rejecting reasonable +// requests on smaller models. +const LOCAL_DEFAULT_COST = { input: 0, output: 0 }; +const LOCAL_DEFAULT_CONTEXT = 32_768; +const LOCAL_DEFAULT_MAX_TOKENS = 8192; + +type LocalOpenAiCompatProvider = "ollama" | "lm-studio"; + +function isLocalOpenAiCompatProvider(provider: string): provider is LocalOpenAiCompatProvider { + return provider === "ollama" || provider === "lm-studio"; +} + +function buildFireworksModel( + modelId: string, + name?: string, + contextWindow?: number, + cost?: { input: number; output: number }, +): Model<"openai-completions"> { + const modelCost = cost ?? FIREWORKS_FALLBACK_COST; return { id: modelId, name: name ?? modelId, @@ -54,19 +83,49 @@ function buildFireworksModel(modelId: string, name?: string, contextWindow?: num baseUrl: FIREWORKS_OPENAI_BASE_URL, reasoning: false, input: ["text"], - cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, + cost: { input: modelCost.input, output: modelCost.output, cacheRead: 0, cacheWrite: 0 }, contextWindow: contextWindow ?? FIREWORKS_DEFAULT_CONTEXT, maxTokens: FIREWORKS_DEFAULT_MAX_TOKENS, }; } -export function resolveModel(provider: string, modelId: string): Model | null { +function buildLocalModel( + provider: LocalOpenAiCompatProvider, + modelId: string, + baseUrl: string, +): Model<"openai-completions"> { + return { + id: modelId, + name: modelId, + api: "openai-completions", + provider, + baseUrl, + reasoning: false, + input: ["text"], + cost: { input: LOCAL_DEFAULT_COST.input, output: LOCAL_DEFAULT_COST.output, cacheRead: 0, cacheWrite: 0 }, + contextWindow: LOCAL_DEFAULT_CONTEXT, + maxTokens: LOCAL_DEFAULT_MAX_TOKENS, + }; +} + +export function resolveModel( + provider: string, + modelId: string, + options?: { baseUrl?: string }, +): Model | null { if (provider === "fireworks-firepass") { - return buildFireworksModel(FIREWORKS_FIREPASS_MODEL_ID, FIREWORKS_FIREPASS_MODEL_NAME, 262_144); + return buildFireworksModel(FIREWORKS_FIREPASS_MODEL_ID, FIREWORKS_FIREPASS_MODEL_NAME, 262_144, FIREWORKS_FIREPASS_COST); } if (provider === "fireworks") { const curated = FIREWORKS_CURATED_MODELS.find((entry) => entry.id === modelId); - return buildFireworksModel(modelId, curated?.name, curated?.contextWindow); + return buildFireworksModel(modelId, curated?.name, curated?.contextWindow, curated?.cost); + } + if (isLocalOpenAiCompatProvider(provider)) { + if (!options?.baseUrl) { + log.warn(`[model-resolver] ${provider} requires a baseUrl from credentials; none provided.`); + return null; + } + return buildLocalModel(provider, modelId, options.baseUrl); } if (!isPiKnownProvider(provider)) { log.warn(`[model-resolver] Unknown pi provider: ${provider}`); @@ -80,6 +139,49 @@ export function resolveModel(provider: string, modelId: string): Model | nu } } +const CROSS_PROVIDER_COST_LOOKUP: Record = { + "github-copilot": ["anthropic", "openai", "google", "xai"], + "google-gemini-cli": ["google"], +}; + +const PROVIDER_FALLBACK_COST: Record = { + "zai": { input: 1, output: 3.2 }, + "kimi-coding": { input: 0.6, output: 3 }, +}; + +function normalizeModelId(id: string): string { + return id.replace(/\./g, "-"); +} + +export function resolveModelCostFallback(model: Model, provider: string): Model { + if (model.cost.input > 0 || model.cost.output > 0) return model; + + const crossProviders = CROSS_PROVIDER_COST_LOOKUP[provider]; + if (crossProviders) { + const normalized = normalizeModelId(model.id); + for (const cp of crossProviders) { + try { + const sourceModels = getModels(cp as Parameters[0]); + const match = sourceModels.find((m) => m.id === model.id) + ?? sourceModels.find((m) => normalizeModelId(m.id) === normalized) + ?? sourceModels.find((m) => normalized.includes(normalizeModelId(m.id)) || normalizeModelId(m.id).includes(normalized)); + if (match && (match.cost.input > 0 || match.cost.output > 0)) { + return { ...model, cost: { ...match.cost } }; + } + } catch { + continue; + } + } + } + + const fallback = PROVIDER_FALLBACK_COST[provider]; + if (fallback) { + return { ...model, cost: { ...model.cost, input: fallback.input, output: fallback.output } }; + } + + return model; +} + export function resolveModels(provider: string): Array<{ id: string; name: string }> { if (provider === "fireworks-firepass") { return [{ id: FIREWORKS_FIREPASS_MODEL_ID, name: FIREWORKS_FIREPASS_MODEL_NAME }]; @@ -87,6 +189,13 @@ export function resolveModels(provider: string): Array<{ id: string; name: strin if (provider === "fireworks") { return FIREWORKS_CURATED_MODELS.map(({ id, name }) => ({ id, name })); } + if (isLocalOpenAiCompatProvider(provider)) { + // Local-server catalogs depend on what the user has loaded and must be + // fetched against the saved baseUrl. The sync resolver returns an + // empty list; the renderer reaches the live catalog through + // `listLocalConnectorModels` in `main-runtime-support.ts`. + return []; + } if (!isPiKnownProvider(provider)) { log.warn(`[model-resolver] Unknown pi provider: ${provider}`); return []; @@ -100,6 +209,37 @@ export function resolveModels(provider: string): Array<{ id: string; name: strin } } +/** + * Fetch the live model list from a local OpenAI-compatible server. + * + * Both Ollama (`/v1/models`) and LM Studio (`/v1/models`) expose the + * standard OpenAI catalog shape — `{ data: [{ id, ...}] }` — so a single + * fetch path handles both. Returns an empty list when the server is + * unreachable so the renderer can surface a "not running" message + * instead of throwing. + */ +export async function fetchLocalConnectorModels( + baseUrl: string, +): Promise> { + const url = `${baseUrl.replace(/\/$/, "")}/models`; + try { + const response = await fetch(url, { method: "GET" }); + if (!response.ok) { + log.warn(`[model-resolver] local model list ${url} returned HTTP ${response.status}`); + return []; + } + const body = (await response.json()) as { data?: Array<{ id?: unknown }> }; + if (!Array.isArray(body.data)) return []; + return body.data + .map((row) => (typeof row?.id === "string" ? row.id : null)) + .filter((id): id is string => id !== null) + .map((id) => ({ id, name: id })); + } catch (err) { + log.warn(`[model-resolver] fetch ${url} failed:`, err instanceof Error ? err.message : String(err)); + return []; + } +} + /** * Extra HTTP headers that must be merged into the streamSimple call for the * given provider. Returns undefined when no extra headers are required. diff --git a/src/main/orchestration/heartbeat.ts b/src/main/orchestration/heartbeat.ts index e68c6660..fb2475e9 100644 --- a/src/main/orchestration/heartbeat.ts +++ b/src/main/orchestration/heartbeat.ts @@ -196,6 +196,8 @@ export function createHeartbeat( return BLOCKED("policy_blocked", "wake_on_demand_disabled"); } if (db.core.agents.isAgentPendingApproval(agentId)) return BLOCKED("human_blocked", "pending_approval"); + const agentLevelBlock = db.core.approvals.getBlockingApproval(companyId, agentId, null); + if (agentLevelBlock) return BLOCKED("human_blocked", `approval_gate:${agentLevelBlock.type}`); const companyMode = db.core.company.getCompanyExecutionMode(companyId); const approvalSettings = db.core.company.getCompanyApprovalSettings(companyId); @@ -268,6 +270,11 @@ export function createHeartbeat( logger.info(`[heartbeat] Starting taskless heartbeat run for ${agent.name}`); } + if (task) { + const taskBlock = db.core.approvals.getBlockingApproval(companyId, agent.id, task.id); + if (taskBlock) return SKIPPED(BLOCKED("human_blocked", `approval_gate:${taskBlock.type}`)); + } + let workspaceResolution: ReturnType; try { workspaceResolution = resolveWorkspace(task, agent, companyId); diff --git a/src/main/orchestration/scheduler.ts b/src/main/orchestration/scheduler.ts index 458828b5..d10a9a4c 100644 --- a/src/main/orchestration/scheduler.ts +++ b/src/main/orchestration/scheduler.ts @@ -223,11 +223,12 @@ export function createScheduler( if (hasActiveRun({ agentId: ceo.id, companyId: company.id })) return; const lastHeartbeat = ceo.lastHeartbeatAt ? new Date(ceo.lastHeartbeatAt).getTime() : 0; - const lastAttempt = bootstrapAttemptByAgent.get(ceo.id) ?? 0; + const attemptKey = agentAttemptKey(company.id, ceo.id); + const lastAttempt = bootstrapAttemptByAgent.get(attemptKey) ?? 0; const bootstrapCooldown = 120_000; if (now - Math.max(lastHeartbeat, lastAttempt) <= bootstrapCooldown) return; - bootstrapAttemptByAgent.set(ceo.id, now); + bootstrapAttemptByAgent.set(attemptKey, now); logger.info(`[scheduler] Company ${company.name} has ${goalCount} goal(s) and no advancing tasks; bootstrapping CEO ${ceo.name}`); wakeAgentIfPossible(ceo.id, company.id, "bootstrap"); } @@ -243,6 +244,8 @@ export function createScheduler( const wokenAgentIds = new Set(); const pendingWakeupAgents = db.automation.wakeup.listPendingWakeupAgents(companyId); for (const { agentId } of pendingWakeupAgents) { + // Queued wakeups are explicit durable events; they bypass the periodic + // heartbeat_enabled scan and let wakeAgentIfPossible own re-enabling. if (hasActiveRun({ agentId, companyId })) continue; if (deps.checkpointEngine?.shouldPauseHeartbeat(companyId, agentId)) continue; @@ -264,21 +267,22 @@ export function createScheduler( } function runAssignedTaskWakeCadence(companyId: string, now: number, wokenAgentIds: Set) { - const activeAgents = db.core.agents.listActiveAgentsForCompany(companyId); + const activeAgents = db.core.agents.listHeartbeatEligibleAgents(companyId); for (const activeAgent of activeAgents) { if (wokenAgentIds.has(activeAgent.id)) continue; if (hasActiveRun({ agentId: activeAgent.id, companyId })) continue; if (deps.checkpointEngine?.shouldPauseHeartbeat(companyId, activeAgent.id)) continue; const lastHeartbeat = activeAgent.lastHeartbeatAt ? new Date(activeAgent.lastHeartbeatAt).getTime() : 0; - const lastAttempt = taskWakeAttemptByAgent.get(activeAgent.id) ?? 0; - const intervalMs = (activeAgent.heartbeatIntervalSec || 120) * 1000; + const attemptKey = agentAttemptKey(companyId, activeAgent.id); + const lastAttempt = taskWakeAttemptByAgent.get(attemptKey) ?? 0; + const intervalMs = (activeAgent.heartbeatIntervalSec ?? 120) * 1000; if (now - Math.max(lastHeartbeat, lastAttempt) < intervalMs) continue; const actionableTasks = db.core.tasks.listAgentTasks(activeAgent.id, companyId, ["todo"]); if (!actionableTasks.some((task) => !task.activeRunId)) continue; - taskWakeAttemptByAgent.set(activeAgent.id, now); + taskWakeAttemptByAgent.set(attemptKey, now); wakeAgentIfPossible(activeAgent.id, companyId, "timer"); } } diff --git a/src/main/orchestration/task-runner.ts b/src/main/orchestration/task-runner.ts index 8601ff75..16f9d6fd 100644 --- a/src/main/orchestration/task-runner.ts +++ b/src/main/orchestration/task-runner.ts @@ -214,6 +214,31 @@ export function createTaskRunner( } const snapshot = db.projection.snapshot.listSnapshot(); const companiesById = new Map(snapshot.companies.map((company) => [company.id, company])); + const reservedAgentIds = new Set(); + const reservedTaskIds = new Set(); + + function finishQueuedRecoveryRun( + run: ReturnType[number], + status: "cancelled" | "interrupted", + summary: string, + errorMessage: string | null, + ): void { + db.core.runs.finishRun({ + runId: run.id, + status, + summary, + errorMessage, + exitCode: null, + signal: null, + model: null, + unattributedCost: true, + recoverTaskForRetry: run.taskId !== null, + }); + if (run.workspaceId) { + db.core.workspaces.releaseWorkspaceLock(run.workspaceId, run.id); + } + emitEvent({ type: "run-updated", runId: run.id, action: "completed" }); + } for (const run of db.core.runs.listQueuedRuns()) { const company = companiesById.get(run.companyId); @@ -221,13 +246,23 @@ export function createTaskRunner( continue; } - const hasBlockingSiblingRun = snapshot.runs.some((entry) => + const hasRunningSiblingRun = snapshot.runs.some((entry) => entry.id !== run.id && entry.companyId === run.companyId && - ["queued", "running"].includes(entry.status) && + entry.status === "running" && (entry.taskId === run.taskId || entry.agentId === run.agentId), ); - if (hasBlockingSiblingRun) { + if (hasRunningSiblingRun) { + continue; + } + + if (reservedAgentIds.has(run.agentId) || (run.taskId !== null && reservedTaskIds.has(run.taskId))) { + finishQueuedRecoveryRun( + run, + "cancelled", + "Queued run superseded during recovery.", + null, + ); continue; } @@ -235,6 +270,24 @@ export function createTaskRunner( const task = run.taskId ? db.core.entity.getTask(run.taskId) : null; const agent = db.core.entity.getAgent(run.agentId); const connector = db.core.connectors.getConnector(run.connectorId); + const blockingApproval = db.core.approvals.getBlockingApproval(run.companyId, agent.id, run.taskId); + if (blockingApproval) { + finishQueuedRecoveryRun( + run, + "cancelled", + `Queued run blocked by pending approval: ${blockingApproval.type}`, + null, + ); + db.core.activity.addActivity({ + companyId: run.companyId, + actor: "system", + action: "run.recovery_blocked_approval", + entityType: "run", + entityId: run.id, + detail: `Recovered queued run blocked by ${blockingApproval.type} for task "${task?.title ?? "(heartbeat)"}"`, + }); + continue; + } const eligibilityFailure = getRunEligibilityFailure(agent, connector, task?.projectId ?? null); if (eligibilityFailure) { if (eligibilityFailure === "CONNECTOR_MODEL_NOT_ALLOWED") { @@ -257,6 +310,21 @@ export function createTaskRunner( detail: `Recovered queued run blocked by connector/model policy for task "${task?.title ?? "(heartbeat)"}"`, }); emitEvent({ type: "run-updated", runId: run.id, action: "completed" }); + } else { + finishQueuedRecoveryRun( + run, + "cancelled", + "Queued run recovery deferred by eligibility gate.", + eligibilityFailure, + ); + db.core.activity.addActivity({ + companyId: run.companyId, + actor: "system", + action: "run.recovery_deferred", + entityType: "run", + entityId: run.id, + detail: `Recovered queued run deferred by ${eligibilityFailure} for task "${task?.title ?? "(heartbeat)"}"`, + }); } continue; } @@ -275,8 +343,19 @@ export function createTaskRunner( snapshot, trigger: "recovery", }); + const currentRun = db.core.runs.getRun(run.id); + if (currentRun.status === "queued" || currentRun.status === "running") { + reservedAgentIds.add(run.agentId); + if (run.taskId !== null) reservedTaskIds.add(run.taskId); + } } catch (error) { logger.warn(`[run] Skipped queued run recovery for ${run.id}: ${error instanceof Error ? error.message : String(error)}`); + finishQueuedRecoveryRun( + run, + "interrupted", + "Queued run recovery failed.", + error instanceof Error ? error.message : String(error), + ); } } } diff --git a/src/main/prompt-builder/messages/operating-en.ts b/src/main/prompt-builder/messages/operating-en.ts index 11e31cd2..c901c9d6 100644 --- a/src/main/prompt-builder/messages/operating-en.ts +++ b/src/main/prompt-builder/messages/operating-en.ts @@ -39,7 +39,7 @@ You are a middle manager running a team within an automated company. You take re ## Priority 1: Standup & Team Assessment (EVERY wake — SOP-001) 1. GET /api/agents/me/standup — get your full briefing -2. **CHECK MESSAGES FIRST**: GET /api/companies/${companyId}/messages?unreadOnly=true&limit=20 — read and respond to ALL unread messages before doing anything else +2. **CHECK MESSAGES FIRST**: GET /api/companies/${companyId}/messages?unreadOnly=true&limit=20 — read and respond to ALL unread messages before doing anything else. Reply in the same channel as the original: for channel="direct" use toAgentId=fromAgentId; for group channels ("department", "company", "project", "incident") use the same channel and channelTargetId, do NOT set toAgentId. Always set parentMessageId to the original message's id 3. Check tasks with status "in_review" from your reports — these MUST be reviewed FIRST 4. Check for "blocked" tasks from your reports — second priority 5. Read new comments on your tasks since last run @@ -119,7 +119,7 @@ You are an individual contributor — a skilled specialist in an automated compa ## Priority 1: Check for Feedback & New Assignments (EVERY wake — SOP-001) 1. GET /api/agents/me/standup — get your full briefing in ONE call. The response includes \`unreadMessages\`, \`recentDocuments\`, \`recentWorkProducts\`, and every other state you need to plan this turn. -2. **MESSAGES FIRST**: walk \`unreadMessages\` from the standup. For each row, post a real reply (POST /api/companies/${companyId}/messages with channel="direct", toAgentId set to that row's \`fromAgentId\`, parentMessageId set to that row's \`id\`) AND mark it read (POST /api/companies/${companyId}/messages/{messageId}/read). Status comments do not count as replies. +2. **MESSAGES FIRST**: walk \`unreadMessages\` from the standup. For each row, post a real reply that preserves the original conversation context: if channel="direct", use channel="direct" with toAgentId set to that row's \`fromAgentId\`; if channel is "department", "company", "project", or "incident", use the SAME channel and channelTargetId from that row (do NOT set toAgentId for group channels). Always set parentMessageId to that row's \`id\`. Mark it read (POST /api/companies/${companyId}/messages/{messageId}/read). Status comments do not count as replies. 3. Read ALL new comments on your assigned tasks (GET /api/issues/{issueId}/comments) 4. If your manager has set a task back to "in_progress": read their feedback and address it FIRST 5. If there are new comments with questions: respond promptly via POST /api/issues/{issueId}/comments diff --git a/src/main/prompt-builder/wake-reasons/en.ts b/src/main/prompt-builder/wake-reasons/en.ts index 467f0452..5ef6d74e 100644 --- a/src/main/prompt-builder/wake-reasons/en.ts +++ b/src/main/prompt-builder/wake-reasons/en.ts @@ -123,13 +123,13 @@ export const WAKE_REASONS: Record = { ], }, message: { - title: "You received a direct message from another agent.", + title: "You received a message from another agent or the Board.", steps: (v) => [ `GET /api/companies/${v.companyId}/messages?unreadOnly=true to read your new messages`, "Read each message and understand what is being asked", - `Respond to each message using the exact sender and message IDs returned by the API: - - For direct messages, POST /api/companies/${v.companyId}/messages with channel="direct", toAgentId set to the original message's fromAgentId, and parentMessageId set to that message's id - - If the sender is Board, the toAgentId is "__board__"`, + `Respond to each message by preserving the original conversation context: + - For direct messages (channel="direct"): POST /api/companies/${v.companyId}/messages with channel="direct", toAgentId set to the original message's fromAgentId, and parentMessageId set to that message's id. If the sender is Board, set toAgentId to "__board__" + - For group channels (channel="department", "company", "project", or "incident"): POST /api/companies/${v.companyId}/messages with the SAME channel and channelTargetId as the original message, and parentMessageId set to that message's id. Do NOT set toAgentId for group channel replies`, `Mark messages as read: POST /api/companies/${v.companyId}/messages/{messageId}/read`, "If the message requires you to take action on a task, do so", ], diff --git a/src/main/worker-service.ts b/src/main/worker-service.ts index 700d355a..18f1ba5f 100644 --- a/src/main/worker-service.ts +++ b/src/main/worker-service.ts @@ -1,6 +1,6 @@ import { app, utilityProcess } from "electron"; import { workerMessageSchema, type WorkerServiceDependencies } from "./worker/entry/types"; -import { handleWorkerExit } from "./worker/entry/crash-recovery"; +import { handleWorkerExit, requeueQueuedRunsAfterWorkerRestart } from "./worker/entry/crash-recovery"; import { handleWorkerMessage } from "./worker/entry/message-handler"; import { DEFAULT_MAX_CONSECUTIVE_SELF_WAKES } from "./run-attempt-policy"; @@ -44,7 +44,7 @@ export function createWorkerService({ const HEALTHY_UPTIME_MS = 60_000; const crashLoopState = { consecutive: 0, halted: false, lastCreatedAt: 0 }; - function scheduleWorkerRespawn() { + function scheduleWorkerRespawn(afterRespawn?: () => void) { if (crashLoopState.halted || destroyed || isQuitting()) return; const uptimeMs = Date.now() - crashLoopState.lastCreatedAt; if (uptimeMs >= HEALTHY_UPTIME_MS) { @@ -70,6 +70,7 @@ export function createWorkerService({ setTimeout(() => { if (destroyed || isQuitting() || crashLoopState.halted) return; createWorker(); + afterRespawn?.(); }, delay); } @@ -91,29 +92,36 @@ export function createWorkerService({ }, }); - worker.stdout?.on("data", (chunk) => { + const currentWorker = worker; + + currentWorker.stdout?.on("data", (chunk) => { const text = String(chunk).trim(); if (text) logger.debug(`[worker:stdout] ${text}`); }); - worker.stderr?.on("data", (chunk) => { + currentWorker.stderr?.on("data", (chunk) => { const text = String(chunk).trim(); if (text) logger.warn(`[worker:stderr] ${text}`); }); - worker.on("exit", (code) => { + currentWorker.on("exit", (code) => { + if (worker === currentWorker) worker = null; if (isQuitting() || destroyed) return; - handleWorkerExit( + const requeueCandidates = handleWorkerExit( code, - { db, logger, isQuitting, publishDomainChanged, emitEvent, wakeAgentIfPossible, dispatchRunToWorker }, - scheduleWorkerRespawn, - getWorker, + { db, logger, isQuitting, publishDomainChanged, emitEvent, wakeAgentIfPossible }, + ); + scheduleWorkerRespawn( + () => requeueQueuedRunsAfterWorkerRestart( + requeueCandidates, + { db, logger, dispatchRunToWorker }, + ), ); }); - worker.on("message", async (message) => { + currentWorker.on("message", async (message) => { if (isShutdownReadyMessage(message)) { try { - worker?.postMessage({ type: "shutdown-exit" }); + currentWorker.postMessage({ type: "shutdown-exit" }); } catch (error) { logger.debug( `[worker-service] shutdown-exit postMessage failed: ${error instanceof Error ? error.message : String(error)}`, @@ -123,7 +131,7 @@ export function createWorkerService({ } await handleWorkerMessage( message, - worker, + currentWorker, { db, logger, diff --git a/src/main/worker/entry/crash-recovery.ts b/src/main/worker/entry/crash-recovery.ts index 7dcb1e80..5c0cae21 100644 --- a/src/main/worker/entry/crash-recovery.ts +++ b/src/main/worker/entry/crash-recovery.ts @@ -1,6 +1,15 @@ import type { WorkerServiceDependencies } from "./types"; import { resolveWorkspaceWithLockMode } from "../../orchestration/resolve-workspace"; +export interface QueuedRunRecoveryCandidate { + id: string; + taskId: string | null; + agentId: string; + companyId: string; + workspaceId: string | null; + connectorId: string; +} + export function handleWorkerExit( code: number | null, { @@ -10,26 +19,16 @@ export function handleWorkerExit( publishDomainChanged, emitEvent: _emitEvent, wakeAgentIfPossible, - dispatchRunToWorker, - }: Pick, - createWorker: () => void, - getWorker: () => ReturnType | null, -) { - if (isQuitting()) return; + }: Pick, +): QueuedRunRecoveryCandidate[] { + if (isQuitting()) return []; logger.warn(`[worker] Worker exited with code ${code}, recovering...`); const snapshot = db.projection.snapshot.listSnapshot(); const affectedAgentIds = new Set(); const affectedCompanyIds = new Map(); - const requeueCandidates: Array<{ - id: string; - taskId: string | null; - agentId: string; - companyId: string; - workspaceId: string | null; - connectorId: string; - }> = []; + const requeueCandidates: QueuedRunRecoveryCandidate[] = []; for (const run of snapshot.runs) { if (run.status === "running") { @@ -42,14 +41,11 @@ export function handleWorkerExit( signal: null, model: null, unattributedCost: true, + recoverTaskForRetry: run.taskId !== null, }); if (run.workspaceId) { db.core.workspaces.releaseWorkspaceLock(run.workspaceId, run.id); } - // EXEC-02: Release execution lock on crash recovery - if (run.taskId) { - db.core.tasks.releaseTaskLock(run.taskId, run.id); - } affectedAgentIds.add(run.agentId); affectedCompanyIds.set(run.agentId, run.companyId); @@ -77,61 +73,70 @@ export function handleWorkerExit( logger.info(`[worker] Promoted ${promoted} deferred wake(s) for agent ${agentId} after crash recovery`); } const pendingWakeTrigger = db.automation.wakeup.getPreferredWakeTrigger(agentId, cid); - if (pendingWakeTrigger) { - wakeAgentIfPossible(agentId, cid, pendingWakeTrigger); - } + wakeAgentIfPossible(agentId, cid, pendingWakeTrigger ?? "recovery"); } publishDomainChanged(); - createWorker(); + return requeueCandidates; +} - const worker = getWorker(); - if (requeueCandidates.length > 0 && worker) { - logger.info(`[worker] Re-enqueuing ${requeueCandidates.length} queued run(s) after worker restart.`); - const recoverySnapshot = db.projection.snapshot.listSnapshot(); - for (const candidate of requeueCandidates) { - try { - // The crashed run may still own a workspace lock acquired during the - // original dispatch; release it before re-dispatch so the replay - // does not self-collide with "Workspace is locked by another run". - if (candidate.workspaceId) { - db.core.workspaces.releaseWorkspaceLock(candidate.workspaceId, candidate.id); - } - const task = candidate.taskId ? db.core.entity.getTask(candidate.taskId) : null; - const agent = db.core.entity.getAgent(candidate.agentId); - const connector = db.core.connectors.getConnector(agent.connectorId); - const { workspace, lockMode } = resolveWorkspaceWithLockMode(db, task, agent, candidate.companyId); - const recoveryCompany = recoverySnapshot.companies.find((company) => company.id === candidate.companyId); +export function requeueQueuedRunsAfterWorkerRestart( + requeueCandidates: QueuedRunRecoveryCandidate[], + { + db, + logger, + dispatchRunToWorker, + }: Pick, +): void { + if (requeueCandidates.length === 0) return; + logger.info(`[worker] Re-enqueuing ${requeueCandidates.length} queued run(s) after worker restart.`); + const recoverySnapshot = db.projection.snapshot.listSnapshot(); + for (const candidate of requeueCandidates) { + try { + const currentRun = db.core.runs.getRun(candidate.id); + if (currentRun.status !== "queued") continue; + // The crashed run may still own a workspace lock acquired during the + // original dispatch; release it before re-dispatch so the replay + // does not self-collide with "Workspace is locked by another run". + if (candidate.workspaceId) { + db.core.workspaces.releaseWorkspaceLock(candidate.workspaceId, candidate.id); + } + const task = candidate.taskId ? db.core.entity.getTask(candidate.taskId) : null; + const agent = db.core.entity.getAgent(candidate.agentId); + const connector = db.core.connectors.getConnector(agent.connectorId); + const { workspace, lockMode, observationPath } = resolveWorkspaceWithLockMode(db, task, agent, candidate.companyId); + const recoveryCompany = recoverySnapshot.companies.find((company) => company.id === candidate.companyId); - dispatchRunToWorker({ - runId: candidate.id, - companyId: candidate.companyId, - agent, - task, - connector, - workspace, - workspaceMode: lockMode, - companyName: recoveryCompany?.name ?? "Company", - companyDescription: recoveryCompany?.description ?? "", - snapshot: recoverySnapshot, - trigger: "recovery", - }); - } catch (error) { - logger.warn(`[worker] Failed to re-enqueue run ${candidate.id}: ${error instanceof Error ? error.message : String(error)}`); - db.core.runs.finishRun({ - runId: candidate.id, - status: "interrupted", - summary: "Could not re-enqueue after worker restart.", - errorMessage: error instanceof Error ? error.message : String(error), - exitCode: null, - signal: null, - model: null, - unattributedCost: true, - }); - if (candidate.workspaceId) { - db.core.workspaces.releaseWorkspaceLock(candidate.workspaceId, candidate.id); - } + dispatchRunToWorker({ + runId: candidate.id, + companyId: candidate.companyId, + agent, + task, + connector, + workspace, + workspaceMode: lockMode, + observationPath, + companyName: recoveryCompany?.name ?? "Company", + companyDescription: recoveryCompany?.description ?? "", + snapshot: recoverySnapshot, + trigger: "recovery", + }); + } catch (error) { + logger.warn(`[worker] Failed to re-enqueue run ${candidate.id}: ${error instanceof Error ? error.message : String(error)}`); + db.core.runs.finishRun({ + runId: candidate.id, + status: "interrupted", + summary: "Could not re-enqueue after worker restart.", + errorMessage: error instanceof Error ? error.message : String(error), + exitCode: null, + signal: null, + model: null, + unattributedCost: true, + recoverTaskForRetry: candidate.taskId !== null, + }); + if (candidate.workspaceId) { + db.core.workspaces.releaseWorkspaceLock(candidate.workspaceId, candidate.id); } } } diff --git a/src/main/worker/entry/message-handler.ts b/src/main/worker/entry/message-handler.ts index 258ecf89..5c851a3a 100644 --- a/src/main/worker/entry/message-handler.ts +++ b/src/main/worker/entry/message-handler.ts @@ -60,14 +60,26 @@ export async function handleWorkerMessage( deps.getRuntimeBridge?.()?.releaseRunResources(event.runId) ?? { ok: true }; return runtimeReleaseResult; }; + // The worker's per-run state machine releases `active[runId]` only + // after receiving run-completion-ack. Any uncaught throw between + // run-finished and posting the ack permanently leaks the agent — + // every subsequent wake then enqueues behind a queue whose + // `isAgentBusy` check refuses to start it. The ack must therefore + // be posted unconditionally. try { - completion = handleRunFinished(event, deps, context, { - beforePostCompletionAutomation: releaseRuntimeResources, - }); - } finally { - releaseRuntimeResources(); + try { + completion = handleRunFinished(event, deps, context, { + beforePostCompletionAutomation: releaseRuntimeResources, + }); + } finally { + releaseRuntimeResources(); + } + releaseCompletedRunWorkspaceLock(completion, deps); + } catch (error) { + deps.logger.error( + `[worker] run-finished post-handling failed for run ${event.runId}: ${error instanceof Error ? error.message : String(error)}`, + ); } - releaseCompletedRunWorkspaceLock(completion, deps); if (!worker) { throw new Error(`run-completion-ack worker missing for run ${event.runId}`); } diff --git a/src/main/worker/entry/types.ts b/src/main/worker/entry/types.ts index cbc391ba..d7c87177 100644 --- a/src/main/worker/entry/types.ts +++ b/src/main/worker/entry/types.ts @@ -5,6 +5,7 @@ import type { NotificationOptions } from "../../notification-service"; import type { createBudgetEnforcementService } from "../../budget-enforcement-service"; import type { createSessionManagementService } from "../../session-management-service"; import type { RuntimeBridge, RuntimeMessageInjector } from "../../agent-runtime"; +import type { RunDispatchParams } from "../../orchestration/turn-context"; const runStatusSchema = z.enum(["queued", "running", "succeeded", "failed", "cancelled", "timed_out", "interrupted"]); const runUsageStatusSchema = z.enum(RUN_USAGE_STATUSES); @@ -110,19 +111,7 @@ export interface WorkerServiceDependencies { publishDomainChanged: () => void; emitEvent: (event: DesktopEvent) => void; wakeAgentIfPossible: (agentId: string | null | undefined, companyId: string, trigger: string) => string | null; - dispatchRunToWorker: (params: { - runId: string; - companyId: string; - agent: ReturnType; - task: ReturnType | null; - connector: ReturnType; - workspace: ReturnType; - workspaceMode: "exclusive" | "shared" | "none"; - companyName: string; - companyDescription: string; - snapshot: ReturnType; - trigger: string; - }) => void; + dispatchRunToWorker: (params: RunDispatchParams) => void; processRunCompletion: (run: RunRecord, event: z.infer & { type: "run-finished" }) => void; onRunStarted?: (runId: string, taskId: string | null, agentId: string, companyId: string) => void; onRunCompleted?: (runId: string, taskId: string | null, agentId: string, companyId: string, status: string, summary: string | null, errorMessage: string | null) => void; diff --git a/src/main/worker/handlers/run-finished-handler.ts b/src/main/worker/handlers/run-finished-handler.ts index 9ccd2707..e562e2c5 100644 --- a/src/main/worker/handlers/run-finished-handler.ts +++ b/src/main/worker/handlers/run-finished-handler.ts @@ -15,6 +15,10 @@ export interface RunFinishedHooks { | { ok: false; errorMessage: string }; } +function errorMessage(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} + export function handleRunFinished( event: Extract, deps: HandlerDeps, @@ -37,7 +41,11 @@ export function handleRunFinished( } : event; if (boundaryResult && !boundaryResult.ok) { - deps.db.core.runs.markRunCompletionFailure(event.runId, boundaryResult.errorMessage); + try { + deps.db.core.runs.markRunCompletionFailure(event.runId, boundaryResult.errorMessage); + } catch (error) { + deps.logger.error(`[worker] markRunCompletionFailure failed for artifact boundary ${event.runId}: ${errorMessage(error)}`); + } } emitAndNotifyCompletion(effectiveEvent, deps, rfCtx); const { hadDeferredWakes } = runPostCompletionAutomation(effectiveEvent, deps, context, rfCtx); diff --git a/src/main/worker/handlers/run-finished-helpers.ts b/src/main/worker/handlers/run-finished-helpers.ts index 4b1e9cda..9717f8f8 100644 --- a/src/main/worker/handlers/run-finished-helpers.ts +++ b/src/main/worker/handlers/run-finished-helpers.ts @@ -16,6 +16,22 @@ export interface RunFinishedContext { runOutcome: ReturnType; } +function errorMessage(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} + +function runCompletionSideEffect( + logger: HandlerDeps["logger"], + label: string, + operation: () => void, +): void { + try { + operation(); + } catch (error) { + logger.error(`[worker] ${label} failed after run completion: ${errorMessage(error)}`); + } +} + export function finalizeRun( event: RunFinishedEvent, deps: HandlerDeps, @@ -71,13 +87,15 @@ export function finalizeRun( leaseRenewalMap.delete(event.runId); const runOutcome = db.core.runs.finishRun({ ...event, unattributedCost }); - db.core.activity.addActivity({ - companyId: run.companyId, - actor: run.agentId, - action: `run.${event.status}`, - entityType: "run", - entityId: run.id, - detail: event.summary ?? event.errorMessage ?? "", + runCompletionSideEffect(logger, "run activity logging", () => { + db.core.activity.addActivity({ + companyId: run.companyId, + actor: run.agentId, + action: `run.${event.status}`, + entityType: "run", + entityId: run.id, + detail: event.summary ?? event.errorMessage ?? "", + }); }); return { run, task, runOutcome }; @@ -85,11 +103,13 @@ export function finalizeRun( export function releaseCompletedRunWorkspaceLock( rfCtx: RunFinishedContext | null, - deps: Pick, + deps: Pick, ): void { const workspaceId = rfCtx?.run.workspaceId; if (!workspaceId) return; - deps.db.core.workspaces.releaseWorkspaceLock(workspaceId, rfCtx.run.id); + runCompletionSideEffect(deps.logger, "workspace lock release", () => { + deps.db.core.workspaces.releaseWorkspaceLock(workspaceId, rfCtx.run.id); + }); } export function accumulateAndRotateSession( @@ -120,22 +140,26 @@ export function accumulateAndRotateSession( const session = db.core.tasks.getSessionById(agentRecord.sessionId); if (!session) return; - sessionService.accumulateTokenUsage( - session.id, - event.inputTokens ?? 0, - event.outputTokens ?? 0, - event.runId, - ); - // Re-read after accumulation: evaluateRotation must see the counts - // that include this run, otherwise a run that just crossed the - // threshold waits a full additional cycle to rotate — the exact - // failure mode that let transcripts balloon past the context window. - const updatedSession = db.core.tasks.getSessionById(agentRecord.sessionId); - if (!updatedSession) return; - const rotationResult = sessionService.evaluateRotation(updatedSession); - if (rotationResult.shouldRotate) { - sessionService.rotateSession(updatedSession); - logger.info(`[session] Rotated session for agent ${run.agentId}: ${rotationResult.reason}`); + try { + sessionService.accumulateTokenUsage( + session.id, + event.inputTokens ?? 0, + event.outputTokens ?? 0, + event.runId, + ); + // Re-read after accumulation: evaluateRotation must see the counts + // that include this run, otherwise a run that just crossed the + // threshold waits a full additional cycle to rotate — the exact + // failure mode that let transcripts balloon past the context window. + const updatedSession = db.core.tasks.getSessionById(agentRecord.sessionId); + if (!updatedSession) return; + const rotationResult = sessionService.evaluateRotation(updatedSession); + if (rotationResult.shouldRotate) { + sessionService.rotateSession(updatedSession); + logger.info(`[session] Rotated session for agent ${run.agentId}: ${rotationResult.reason}`); + } + } catch (error) { + logger.error(`[session] accumulation/rotation failed for run ${run.id}: ${errorMessage(error)}`); } } @@ -147,20 +171,26 @@ export function emitAndNotifyCompletion( const { db, emitEvent, notify, logger } = deps; const { run, task, runOutcome } = rfCtx; - emitEvent({ type: "run-updated", runId: run.id, action: "completed" }); - emitEvent({ - type: "run-status", - runId: run.id, - status: event.status, - message: event.summary ?? event.errorMessage ?? event.status, + runCompletionSideEffect(logger, "run-updated event emission", () => { + emitEvent({ type: "run-updated", runId: run.id, action: "completed" }); + }); + runCompletionSideEffect(logger, "run-status event emission", () => { + emitEvent({ + type: "run-status", + runId: run.id, + status: event.status, + message: event.summary ?? event.errorMessage ?? event.status, + }); }); if ((event.status === "failed" || event.status === "timed_out" || event.status === "interrupted") && task) { - notify({ - title: "Task Failed", - body: `${task.title} -- manual intervention needed`, - urgency: "informational", - navigation: { section: "tasks", entityId: task.id }, - batchKey: "failed_runs", + runCompletionSideEffect(logger, "task failure notification", () => { + notify({ + title: "Task Failed", + body: `${task.title} -- manual intervention needed`, + urgency: "informational", + navigation: { section: "tasks", entityId: task.id }, + batchKey: "failed_runs", + }); }); } @@ -175,22 +205,28 @@ export function emitAndNotifyCompletion( } if (updatedAgent && runOutcome.budgetHardStopped) { - db.core.activity.addActivity({ - companyId: updatedAgent.companyId, - actor: "system", - action: "agent.auto_paused", - entityType: "agent", - entityId: updatedAgent.id, - detail: `Budget exhausted at ${updatedAgent.spentMonthlyUsd.toFixed(2)} / ${updatedAgent.budgetMonthlyUsd.toFixed(2)} USD.`, + runCompletionSideEffect(logger, "budget auto-pause activity logging", () => { + db.core.activity.addActivity({ + companyId: updatedAgent.companyId, + actor: "system", + action: "agent.auto_paused", + entityType: "agent", + entityId: updatedAgent.id, + detail: `Budget exhausted at ${updatedAgent.spentMonthlyUsd.toFixed(2)} / ${updatedAgent.budgetMonthlyUsd.toFixed(2)} USD.`, + }); }); - notify({ - title: "Agent Paused", - body: `${updatedAgent.name} paused -- budget limit reached`, - urgency: "informational", - navigation: { section: "agents", entityId: updatedAgent.id }, - batchKey: "agent_paused", + runCompletionSideEffect(logger, "budget auto-pause notification", () => { + notify({ + title: "Agent Paused", + body: `${updatedAgent.name} paused -- budget limit reached`, + urgency: "informational", + navigation: { section: "agents", entityId: updatedAgent.id }, + batchKey: "agent_paused", + }); + }); + runCompletionSideEffect(logger, "agent-updated event emission", () => { + emitEvent({ type: "agent-updated", agentId: updatedAgent.id, action: "paused" }); }); - emitEvent({ type: "agent-updated", agentId: updatedAgent.id, action: "paused" }); } } @@ -203,58 +239,97 @@ export function runPostCompletionAutomation( const { db, logger, isQuitting, wakeAgentIfPossible, processRunCompletion, onRunCompleted, notify, budgetService } = deps; const { destroyed, MAX_RETRY_ATTEMPTS, RETRY_COOLDOWN_MS, pendingRetryTimers } = context; const { run, task } = rfCtx; + let housekeepingStatus = event.status; + let housekeepingSummary = event.summary ?? null; + let housekeepingErrorMessage = event.errorMessage ?? null; // Policy-based incident check after run completion. Always runs once the // run is finalized — cost is read from the cost_entries SUM inside // finishRun/checkAndCreateIncidents, not from the worker event. if (budgetService && event.status === "succeeded") { - budgetService.checkAndCreateIncidents(run.companyId, run.agentId, task?.projectId ?? null); + try { + budgetService.checkAndCreateIncidents(run.companyId, run.agentId, task?.projectId ?? null); + } catch (error) { + logger.error(`[worker] budget incident check failed for run ${run.id}: ${errorMessage(error)}`); + } } if ((event.status === "failed" || event.status === "timed_out") && !isQuitting() && !destroyed && run.taskId) { - handleFailureEscalation( - { - db, - wakeAgentIfPossible, - notify: (title, body) => - notify({ title, body, urgency: "critical", navigation: { section: "tasks", entityId: run.taskId ?? "" } }), - logger, - }, - { taskId: run.taskId, agentId: run.agentId, companyId: run.companyId }, - task, - event.status, - MAX_RETRY_ATTEMPTS, - RETRY_COOLDOWN_MS, - (fn, delayMs) => { - const retryTimer = setTimeout(() => { - pendingRetryTimers.delete(retryTimer); - if (isQuitting() || destroyed) return; - fn(); - }, delayMs); - pendingRetryTimers.add(retryTimer); - }, - ); + try { + handleFailureEscalation( + { + db, + wakeAgentIfPossible, + notify: (title, body) => + notify({ title, body, urgency: "critical", navigation: { section: "tasks", entityId: run.taskId ?? "" } }), + logger, + }, + { taskId: run.taskId, agentId: run.agentId, companyId: run.companyId }, + task, + event.status, + MAX_RETRY_ATTEMPTS, + RETRY_COOLDOWN_MS, + (fn, delayMs) => { + const retryTimer = setTimeout(() => { + pendingRetryTimers.delete(retryTimer); + if (isQuitting() || destroyed) return; + fn(); + }, delayMs); + pendingRetryTimers.add(retryTimer); + }, + ); + } catch (error) { + logger.error(`[worker] failure escalation failed for run ${run.id}: ${errorMessage(error)}`); + } } // EXEC-05: Promote deferred wakes after run completion let hadDeferredWakes = false; if (!isQuitting() && !destroyed) { - const promoted = db.automation.wakeup.promoteWakeupRequests(run.agentId, run.companyId); - const pendingWakeTrigger = db.automation.wakeup.getPreferredWakeTrigger(run.agentId, run.companyId); - if (promoted > 0) { - logger.info(`[worker] Promoted ${promoted} deferred wake(s) for agent ${run.agentId}`); - } - if (pendingWakeTrigger) { - hadDeferredWakes = true; - logger.info( - `[worker] Processing pending wake(s) for agent ${run.agentId} with priority trigger "${pendingWakeTrigger}"`, - ); - wakeAgentIfPossible(run.agentId, run.companyId, pendingWakeTrigger); + try { + const promoted = db.automation.wakeup.promoteWakeupRequests(run.agentId, run.companyId); + const pendingWakeTrigger = db.automation.wakeup.getPreferredWakeTrigger(run.agentId, run.companyId); + if (promoted > 0) { + logger.info(`[worker] Promoted ${promoted} deferred wake(s) for agent ${run.agentId}`); + } + if (pendingWakeTrigger) { + hadDeferredWakes = true; + logger.info( + `[worker] Processing pending wake(s) for agent ${run.agentId} with priority trigger "${pendingWakeTrigger}"`, + ); + wakeAgentIfPossible(run.agentId, run.companyId, pendingWakeTrigger); + } + } catch (error) { + logger.error(`[worker] deferred wake processing failed for run ${run.id}: ${errorMessage(error)}`); } } if (!isQuitting() && !destroyed) { - processRunCompletion(run, event as z.infer & { type: "run-finished" }); + try { + processRunCompletion(run, event as z.infer & { type: "run-finished" }); + } catch (error) { + const message = errorMessage(error); + logger.error(`[worker] processRunCompletion failed for run ${run.id}: ${message}`); + housekeepingStatus = "failed"; + housekeepingSummary = null; + housekeepingErrorMessage = message; + try { + db.core.runs.markRunCompletionFailure(run.id, message); + } catch (markError) { + logger.error(`[worker] markRunCompletionFailure failed for run ${run.id}: ${errorMessage(markError)}`); + } + if (task?.status === "in_progress") { + try { + db.core.tasks.advanceTaskStatus( + task.id, + "blocked", + `Run completion automation failed: ${message}`, + ); + } catch (statusError) { + logger.error(`[worker] failed to block task ${task.id} after completion automation failure: ${errorMessage(statusError)}`); + } + } + } } // Run housekeeping: thread events + knowledge candidates @@ -265,9 +340,9 @@ export function runPostCompletionAutomation( run.taskId, run.agentId, run.companyId, - event.status, - event.summary ?? null, - event.errorMessage ?? null, + housekeepingStatus, + housekeepingSummary, + housekeepingErrorMessage, ); } catch (err) { logger.error(`[worker] onRunCompleted housekeeping failed: ${err instanceof Error ? err.message : String(err)}`); @@ -275,7 +350,7 @@ export function runPostCompletionAutomation( } // Reset retry count on successful completion - if (event.status === "succeeded" && run.taskId) { + if (housekeepingStatus === "succeeded" && run.taskId) { handleSuccessReset(db, run.taskId, logger); } diff --git a/src/main/worker/runtime/index.ts b/src/main/worker/runtime/index.ts index 6c1da392..e9b6888b 100644 --- a/src/main/worker/runtime/index.ts +++ b/src/main/worker/runtime/index.ts @@ -19,6 +19,8 @@ interface PrepareRunResult { provider: string; apiKey: string; extraEnv: Record; + /** OpenAI-compatible server base URL for local connectors (Ollama, LM Studio). */ + baseUrl: string | null; priorMessages: AgentMessage[]; } @@ -133,7 +135,9 @@ export class WorkerRunner { const runEnv = mergeRunEnvs(job.env, prepared.extraEnv); redact = buildRedactor(runEnv); - const model = resolveModel(prepared.provider, job.model); + const model = resolveModel(prepared.provider, job.model, { + baseUrl: prepared.baseUrl ?? undefined, + }); if (!model) { throw new Error(`Unknown model ${job.model} for provider ${prepared.provider}.`); } diff --git a/src/renderer/components/company-onboarding/provider-cards/ProviderCard.tsx b/src/renderer/components/company-onboarding/provider-cards/ProviderCard.tsx index ad49075f..c1a30dc3 100644 --- a/src/renderer/components/company-onboarding/provider-cards/ProviderCard.tsx +++ b/src/renderer/components/company-onboarding/provider-cards/ProviderCard.tsx @@ -4,6 +4,7 @@ import type { TranslationFn } from "../../../i18n"; import { OAuthProviderCard } from "./oauth-card"; import { ApiKeyProviderCard } from "./api-key-card"; import { CloudProviderCard } from "./cloud-card"; +import { LocalProviderCard } from "./local-card"; export interface ProviderCardProps { entry: ConnectorCatalogEntry; @@ -15,8 +16,8 @@ export interface ProviderCardProps { } /** - * Dispatch based on the catalog auth descriptor. Five auth kinds live behind - * three dedicated cards — cloud variants share a card because their onboarding + * Dispatch based on the catalog auth descriptor. Six auth kinds live behind + * four dedicated cards — cloud variants share a card because their onboarding * affordance is identical (deep link to Settings). A universal form generator * is intentionally avoided so each auth flow stays legible end-to-end. */ @@ -31,5 +32,7 @@ export function OnboardingProviderDispatch(props: ProviderCardProps) { case "azure": case "gcp_adc": return ; + case "local": + return ; } } diff --git a/src/renderer/components/company-onboarding/provider-cards/local-card.tsx b/src/renderer/components/company-onboarding/provider-cards/local-card.tsx new file mode 100644 index 00000000..fbab8f41 --- /dev/null +++ b/src/renderer/components/company-onboarding/provider-cards/local-card.tsx @@ -0,0 +1,155 @@ +import { useEffect, useState } from "react"; +import { FloppyDisk } from "../../ui/icons"; +import type { ConnectorCatalogEntry } from "@shared/connector-catalog"; +import type { CredentialSummary } from "@shared/contracts/schemas-credentials"; +import type { TranslationFn } from "../../../i18n"; +import { + InlineFeedback, + OnboardingProviderCard, + OnboardingStatusBadge, + type OnboardingCardStatus, +} from "./shared"; +import { Input } from "../../ui"; + +interface LocalProviderCardProps { + entry: ConnectorCatalogEntry; + credential: CredentialSummary | null; + expanded: boolean; + onToggle: () => void; + t: TranslationFn; +} + +/** + * Onboarding card for local OpenAI-compatible servers (Ollama, LM Studio). + * + * The catalog supplies a sensible default URL per entry; the user can + * override it to point at a non-default port, a remote machine, or a + * reverse proxy. The card never asks for an API key — local servers + * accept any bearer value, so the credential resolver supplies the + * connector id as a non-empty placeholder for the OpenAI SDK. + */ +export function LocalProviderCard({ + entry, + credential, + expanded, + onToggle, + t, +}: LocalProviderCardProps) { + if (entry.auth.type !== "local") { + throw new Error(`LocalProviderCard received non-local connector: ${entry.id}`); + } + const defaultBaseUrl = entry.auth.defaultBaseUrl; + const initialBaseUrl = credential?.displayHint?.startsWith("local:") + ? credential.displayHint.slice("local:".length) + : defaultBaseUrl; + const [baseUrl, setBaseUrl] = useState(initialBaseUrl); + const [busy, setBusy] = useState(false); + const [feedback, setFeedback] = useState(null); + + useEffect(() => { + const next = credential?.displayHint?.startsWith("local:") + ? credential.displayHint.slice("local:".length) + : defaultBaseUrl; + setBaseUrl(next); + }, [credential?.displayHint, defaultBaseUrl]); + + const status: OnboardingCardStatus = busy + ? "checking" + : credential?.status ?? "not_configured"; + const connected = credential?.status === "active"; + + const handleSave = async () => { + if (!window.agentCompany) return; + const trimmed = baseUrl.trim(); + if (trimmed.length === 0) { + setFeedback(t("onboarding.providers.baseUrlRequired")); + return; + } + try { + new URL(trimmed); + } catch { + setFeedback(t("onboarding.providers.baseUrlInvalid")); + return; + } + setBusy(true); + setFeedback(null); + const result = await window.agentCompany.saveCredential({ + connectorId: entry.id, + payload: { + kind: "local", + value: { baseUrl: trimmed }, + }, + }); + setBusy(false); + if (!result.ok) { + setFeedback(result.error.message); + return; + } + setFeedback(t("onboarding.providers.saved")); + }; + + const handleRemove = async () => { + if (!window.agentCompany) return; + setBusy(true); + setFeedback(null); + const result = await window.agentCompany.removeCredential({ connectorId: entry.id }); + setBusy(false); + if (!result.ok) setFeedback(result.error.message); + }; + + return ( + } + expanded={expanded} + onToggle={onToggle} + > + {entry.notes ? ( +
{entry.notes}
+ ) : null} +
{ + event.preventDefault(); + void handleSave(); + }} + className="space-y-3" + > +
+ + +
+
+ + {connected ? ( + + ) : null} +
+
+ +
+ ); +} diff --git a/src/renderer/components/settings/forms/LocalForm.tsx b/src/renderer/components/settings/forms/LocalForm.tsx new file mode 100644 index 00000000..caf28f90 --- /dev/null +++ b/src/renderer/components/settings/forms/LocalForm.tsx @@ -0,0 +1,79 @@ +import { useState } from "react"; +import { ActionButton, Input } from "../../ui"; +import { useT } from "../../../i18n"; + +const desktop = (typeof window !== "undefined" ? window.agentCompany : null) as + | (Window["agentCompany"] | null); + +interface Props { + connectorId: string; + defaultBaseUrl: string; + onSaved?: () => void; +} + +export function LocalForm({ connectorId, defaultBaseUrl, onSaved }: Props) { + const t = useT(); + const [baseUrl, setBaseUrl] = useState(defaultBaseUrl); + const [busy, setBusy] = useState(false); + const [error, setError] = useState(null); + + const trimmed = baseUrl.trim(); + const canSave = trimmed.length > 0 && !busy && Boolean(desktop); + + const handleSubmit = async () => { + if (!desktop || !canSave) return; + try { + new URL(trimmed); + } catch { + setError(t("onboarding.providers.baseUrlInvalid")); + return; + } + setBusy(true); + setError(null); + const result = await desktop.saveCredential({ + connectorId, + payload: { kind: "local", value: { baseUrl: trimmed } }, + }); + setBusy(false); + if (!result.ok) { + setError(result.error.message); + return; + } + onSaved?.(); + }; + + return ( +
{ + event.preventDefault(); + void handleSubmit(); + }} + className="space-y-3" + > + + + {error ? ( +
+ {error} +
+ ) : null} + +
+ +
+
+ ); +} diff --git a/src/renderer/components/settings/providers/ProviderRow.tsx b/src/renderer/components/settings/providers/ProviderRow.tsx index fa8cd697..a716626b 100644 --- a/src/renderer/components/settings/providers/ProviderRow.tsx +++ b/src/renderer/components/settings/providers/ProviderRow.tsx @@ -5,6 +5,7 @@ import { ApiKeyForm } from "../forms/ApiKeyForm"; import { AzureForm } from "../forms/AzureForm"; import { AwsForm } from "../forms/AwsForm"; import { VertexForm } from "../forms/VertexForm"; +import { LocalForm } from "../forms/LocalForm"; import { ActionButton } from "../../ui"; import { useT, type TranslationFn } from "../../../i18n"; @@ -28,6 +29,8 @@ function kindLabel( return t("providers.kindAws"); case "gcp_adc": return t("providers.kindGcp"); + case "local": + return t("providers.kindLocal"); } } @@ -118,6 +121,8 @@ export function ProviderRow({ entry, credential }: Props) { return ; if (entry.auth.type === "gcp_adc") return ; + if (entry.auth.type === "local") + return ; return null; }; diff --git a/src/renderer/i18n/en.ts b/src/renderer/i18n/en.ts index 5b14d459..43578420 100644 --- a/src/renderer/i18n/en.ts +++ b/src/renderer/i18n/en.ts @@ -189,6 +189,9 @@ export const en: Record = { "onboarding.providers.apiKeyLabel": "API key", "onboarding.providers.apiKeyRequired": "Enter an API key before saving.", "onboarding.providers.apiKeyReplacePlaceholder": "Leave empty to keep the current key", + "onboarding.providers.baseUrlLabel": "Base URL", + "onboarding.providers.baseUrlRequired": "Enter the local server URL before saving.", + "onboarding.providers.baseUrlInvalid": "Enter a valid HTTP or HTTPS URL.", "onboarding.providers.save": "Save", "onboarding.providers.saving": "Saving…", "onboarding.providers.saved": "Saved.", @@ -1130,6 +1133,7 @@ export const en: Record = { "providers.kindAzure": "Azure", "providers.kindAws": "AWS", "providers.kindGcp": "Vertex (ADC)", + "providers.kindLocal": "Local", "providers.statusConnected": "Connected", "providers.statusExpired": "Expired", "providers.statusError": "Error", diff --git a/src/renderer/i18n/zh.ts b/src/renderer/i18n/zh.ts index 97f25c7a..692b9272 100644 --- a/src/renderer/i18n/zh.ts +++ b/src/renderer/i18n/zh.ts @@ -189,6 +189,9 @@ export const zh: Record = { "onboarding.providers.apiKeyLabel": "API Key", "onboarding.providers.apiKeyRequired": "请输入 API Key 后再保存。", "onboarding.providers.apiKeyReplacePlaceholder": "保持现有 Key 请留空", + "onboarding.providers.baseUrlLabel": "服务地址", + "onboarding.providers.baseUrlRequired": "请填入本地服务地址后再保存。", + "onboarding.providers.baseUrlInvalid": "请输入合法的 HTTP 或 HTTPS 地址。", "onboarding.providers.save": "保存", "onboarding.providers.saving": "保存中…", "onboarding.providers.saved": "已保存。", @@ -1130,6 +1133,7 @@ export const zh: Record = { "providers.kindAzure": "Azure", "providers.kindAws": "AWS", "providers.kindGcp": "Vertex (ADC)", + "providers.kindLocal": "本地", "providers.statusConnected": "已连接", "providers.statusExpired": "已过期", "providers.statusError": "出错", diff --git a/src/shared/connector-catalog.ts b/src/shared/connector-catalog.ts index 4b46089a..cb7b68ae 100644 --- a/src/shared/connector-catalog.ts +++ b/src/shared/connector-catalog.ts @@ -16,20 +16,22 @@ export type PiOAuthProviderId = | "google-gemini-cli" | "google-antigravity"; -export type ConnectorAuthKind = "oauth" | "api_key" | "aws" | "azure" | "gcp_adc"; +export type ConnectorAuthKind = "oauth" | "api_key" | "aws" | "azure" | "gcp_adc" | "local"; export interface ConnectorAuthOAuth { type: "oauth"; oauthId: PiOAuthProviderId } export interface ConnectorAuthApiKey { type: "api_key"; envKey: string } export interface ConnectorAuthAws { type: "aws" } export interface ConnectorAuthAzure { type: "azure" } export interface ConnectorAuthGcpAdc { type: "gcp_adc" } +export interface ConnectorAuthLocal { type: "local"; defaultBaseUrl: string } export type ConnectorAuth = | ConnectorAuthOAuth | ConnectorAuthApiKey | ConnectorAuthAws | ConnectorAuthAzure - | ConnectorAuthGcpAdc; + | ConnectorAuthGcpAdc + | ConnectorAuthLocal; export interface ConnectorCatalogEntry { id: ConnectorId; @@ -236,9 +238,30 @@ const API_KEY_ENTRIES: ConnectorCatalogEntry[] = [ provider: "fireworks-firepass", label: "Fireworks Fire Pass", description: "Fireworks Fire Pass — subscription bundle. Same FIREWORKS_API_KEY.", - defaultModel: "accounts/fireworks/routers/kimi-k2p5-turbo", + defaultModel: "accounts/fireworks/routers/kimi-k2p6-turbo", auth: { type: "api_key", envKey: "FIREWORKS_API_KEY" }, - notes: "Fire Pass uses the dedicated router id accounts/fireworks/routers/kimi-k2p5-turbo documented at docs.fireworks.ai/firepass.", + notes: "Fire Pass uses the dedicated router id accounts/fireworks/routers/kimi-k2p6-turbo documented at docs.fireworks.ai/firepass.", + }, +]; + +const LOCAL_ENTRIES: ConnectorCatalogEntry[] = [ + { + id: "ollama", + provider: "ollama", + label: "Ollama", + description: "Local Ollama server — open-weights models on your own machine via the OpenAI-compatible endpoint.", + defaultModel: "llama3.2", + auth: { type: "local", defaultBaseUrl: "http://localhost:11434/v1" }, + notes: "Default port 11434. Override if you launched `ollama serve` elsewhere.", + }, + { + id: "lm-studio", + provider: "lm-studio", + label: "LM Studio", + description: "Local LM Studio server — host any loaded model via the OpenAI-compatible endpoint.", + defaultModel: "local-model", + auth: { type: "local", defaultBaseUrl: "http://localhost:1234/v1" }, + notes: "Enable LM Studio's local server from the Developer tab. Default port 1234.", }, ]; @@ -277,6 +300,7 @@ export const CONNECTOR_CATALOG: ConnectorCatalogEntry[] = [ ...OAUTH_ENTRIES, ...API_KEY_ENTRIES, ...CLOUD_ENTRIES, + ...LOCAL_ENTRIES, ]; export function findConnectorCatalogEntry(id: string): ConnectorCatalogEntry | undefined { diff --git a/src/shared/contracts/schemas-company.ts b/src/shared/contracts/schemas-company.ts index 7d3a15ee..75fc2aaa 100644 --- a/src/shared/contracts/schemas-company.ts +++ b/src/shared/contracts/schemas-company.ts @@ -49,6 +49,9 @@ export const connectorIdEnum = z.enum([ "amazon-bedrock", "azure-openai", "google-vertex", + // Local (2) + "ollama", + "lm-studio", ]); export type ConnectorId = z.infer; diff --git a/src/shared/contracts/schemas-credentials.ts b/src/shared/contracts/schemas-credentials.ts index eb91e49f..4d58c5b4 100644 --- a/src/shared/contracts/schemas-credentials.ts +++ b/src/shared/contracts/schemas-credentials.ts @@ -6,6 +6,7 @@ export const credentialKindSchema = z.enum([ "aws", "azure", "gcp_adc", + "local", ]); export const credentialStatusSchema = z.enum([ @@ -49,11 +50,16 @@ export const gcpAdcCredentialSchema = z.object({ credentialsPath: z.string().optional(), }); +export const localCredentialSchema = z.object({ + baseUrl: z.string().url(), +}); + export const credentialPayloadSchema = z.discriminatedUnion("kind", [ z.object({ kind: z.literal("api_key"), value: apiKeyValueSchema }), z.object({ kind: z.literal("azure"), value: azureCredentialSchema }), z.object({ kind: z.literal("aws"), value: awsCredentialSchema }), z.object({ kind: z.literal("gcp_adc"), value: gcpAdcCredentialSchema }), + z.object({ kind: z.literal("local"), value: localCredentialSchema }), ]); export const saveCredentialInputSchema = z.object({ @@ -85,4 +91,5 @@ export type SaveCredentialInput = z.infer; export type AwsCredential = z.infer; export type AzureCredential = z.infer; export type GcpAdcCredential = z.infer; +export type LocalCredential = z.infer; export type ApiKeyValue = z.infer; diff --git a/tests/integration/pi-mono-migrations.spec.ts b/tests/integration/pi-mono-migrations.spec.ts index 4498dd70..c6f3b530 100644 --- a/tests/integration/pi-mono-migrations.spec.ts +++ b/tests/integration/pi-mono-migrations.spec.ts @@ -107,12 +107,14 @@ describe("pi-mono migrations 27 + 28", () => { db.init(); const rows = db.projection.snapshot.listSnapshot().connectors.map((entry) => entry.id); - expect(rows.length).toBe(26); + expect(rows.length).toBe(28); expect(rows).toContain("claude-pro"); expect(rows).toContain("amazon-bedrock"); expect(rows).toContain("google-vertex"); expect(rows).toContain("fireworks"); expect(rows).toContain("fireworks-firepass"); + expect(rows).toContain("ollama"); + expect(rows).toContain("lm-studio"); expect(rows.find((id) => (id as string) === "claude_local")).toBeUndefined(); expect(rows.find((id) => (id as string) === "codex_local")).toBeUndefined(); expect(rows.find((id) => (id as string) === "claude_compatible")).toBeUndefined(); diff --git a/tests/unit/active-run-projection.test.ts b/tests/unit/active-run-projection.test.ts new file mode 100644 index 00000000..b2ce7759 --- /dev/null +++ b/tests/unit/active-run-projection.test.ts @@ -0,0 +1,117 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { AppDatabase } from "../../src/main/database"; + +vi.mock("electron", () => ({ + safeStorage: { + isEncryptionAvailable: () => false, + getSelectedStorageBackend: () => "basic_text", + encryptString: (value: string) => Buffer.from(value, "utf8"), + decryptString: (value: Buffer) => value.toString("utf8"), + }, +})); + +type RawDb = { + db: { + prepare: (sql: string) => { + run: (...args: unknown[]) => void; + }; + }; +}; + +function createCompany(db: AppDatabase): string { + return db.core.company.saveCompany({ + name: "Projection Co", + description: "Active run projection contract", + status: "active", + }); +} + +function createWorkspace(db: AppDatabase, companyId: string, localPath: string): string { + return db.core.workspaces.saveWorkspace({ + companyId, + name: "Primary", + localPath, + repoUrl: "", + repoRef: "", + isPrimary: true, + }); +} + +function createAgent(db: AppDatabase, companyId: string, workspaceId: string): string { + return db.core.agents.saveAgent({ + companyId, + name: "Projection Agent", + role: "Engineer", + title: "Engineer", + status: "idle", + reportsTo: null, + connectorId: "openai-codex", + workspaceId, + model: "gpt-5.4", + capabilities: "Tests active run projections", + budgetMonthlyUsd: 100, + }); +} + +describe("active run projection", () => { + let tempDir: string; + let db: AppDatabase; + + beforeEach(async () => { + tempDir = await mkdtemp(join(tmpdir(), "agentcompany-active-run-projection-")); + db = new AppDatabase(tempDir); + db.init(); + }); + + afterEach(async () => { + db.close(); + await rm(tempDir, { recursive: true, force: true }); + }); + + it("ignores terminal run leases when projecting activeRunId", () => { + const companyId = createCompany(db); + const workspaceId = createWorkspace(db, companyId, tempDir); + const agentId = createAgent(db, companyId, workspaceId); + const taskId = db.core.tasks.saveTask({ + companyId, + title: "Recoverable task", + description: "A stale terminal lease must not hide this task from recovery.", + assigneeAgentId: agentId, + workspaceId, + priority: "high", + status: "todo", + }); + const run = db.core.runs.createRun({ + companyId, + taskId, + agentId, + workspaceId: null, + connectorId: "openai-codex", + }); + db.core.runs.finishRun({ + runId: run.id, + status: "succeeded", + summary: "Done", + errorMessage: null, + exitCode: 0, + signal: null, + model: null, + unattributedCost: true, + }); + const now = new Date().toISOString(); + (db as unknown as RawDb).db + .prepare( + `insert into task_leases ( + task_id, company_id, run_id, agent_id, acquired_at, renewed_at, expires_at, created_at, updated_at + ) values (?, ?, ?, ?, ?, ?, ?, ?, ?)`, + ) + .run(taskId, companyId, run.id, agentId, now, now, now, now, now); + + expect(db.core.entity.getTask(taskId).activeRunId).toBeNull(); + expect(db.core.tasks.listAgentTasks(agentId, companyId, ["todo"])[0]?.activeRunId).toBeNull(); + expect(db.projection.snapshot.listSnapshot({ companyId }).tasks.find((task) => task.id === taskId)?.activeRunId).toBeNull(); + }); +}); diff --git a/tests/unit/approval-review-feedback-writeback.test.ts b/tests/unit/approval-review-feedback-writeback.test.ts index 3aeea474..b0863cdb 100644 --- a/tests/unit/approval-review-feedback-writeback.test.ts +++ b/tests/unit/approval-review-feedback-writeback.test.ts @@ -4,6 +4,7 @@ import { tmpdir } from "node:os"; import { afterEach, describe, expect, it, vi } from "vitest"; import { AppDatabase } from "../../src/main/database"; import { handleApprovalResolved } from "../../src/main/automation/automation-approval-handler"; +import { createHeartbeat } from "../../src/main/orchestration/heartbeat"; const tempDirs: string[] = []; @@ -18,7 +19,9 @@ afterEach(async () => { }); describe("deliverable review feedback writeback", () => { - it("persists high-importance knowledge when the board requests revisions on a deliverable", async () => { + it.each(["rejected", "revision_requested"] as const)( + "persists high-importance knowledge when the board returns %s on a deliverable", + async (decision) => { const profileDir = await createTempDir("agentcompany-review-feedback-"); const db = new AppDatabase(profileDir); db.init(); @@ -87,7 +90,7 @@ describe("deliverable review feedback writeback", () => { db.core.approvals.decideApproval({ approvalId, - state: "revision_requested", + state: decision, decisionNote: "Add stronger competitor evidence and a clearer recommendation.", }); @@ -105,7 +108,7 @@ describe("deliverable review feedback writeback", () => { handleTaskStatusChange: vi.fn(), runWorkflowPipeline: async () => {}, trackTimer: vi.fn(), - }, approvalId, companyId, "revision_requested"); + }, approvalId, companyId, decision); const knowledgeEntries = db.projection.snapshot.listSnapshot().knowledgeBase.filter((entry) => entry.companyId === companyId); expect(knowledgeEntries).toHaveLength(1); @@ -113,13 +116,147 @@ describe("deliverable review feedback writeback", () => { expect(knowledgeEntries[0]?.importance).toBe("high"); expect(knowledgeEntries[0]?.referencedEntityType).toBe("task"); expect(knowledgeEntries[0]?.referencedEntityId).toBe(taskId); - expect(knowledgeEntries[0]?.content).toContain("revision_requested"); + expect(knowledgeEntries[0]?.content).toContain(decision); expect(knowledgeEntries[0]?.content).toContain("stronger competitor evidence"); expect(wakeAgentIfPossible).toHaveBeenCalledWith(agentId, companyId, "revision_needed"); - expect(db.core.entity.getTask(taskId).status).toBe("in_progress"); + expect(wakeAgentIfPossible).toHaveBeenCalledTimes(1); + expect(db.core.entity.getTask(taskId).status).toBe("todo"); const document = db.projection.snapshot.listSnapshot().documents.find((entry) => entry.id === documentId); expect(document?.status).toBe("draft"); + db.close(); + }, + ); + + it("makes a revision request immediately checkoutable by the assignee heartbeat", async () => { + const profileDir = await createTempDir("agentcompany-review-checkout-"); + const db = new AppDatabase(profileDir); + db.init(); + + db.core.connectors.saveConnector({ + id: "openai-codex", + model: "gpt-5.4", + envBindingText: "", + notes: "Test connector", + }); + + const companyId = db.core.company.saveCompany({ + name: "Review Checkout Co", + description: "Tests review revision checkout", + status: "active", + }); + const workspaceId = db.core.workspaces.saveWorkspace({ + companyId, + name: "Primary", + localPath: profileDir, + repoUrl: "", + repoRef: "", + isPrimary: true, + }); + const agentId = db.core.agents.saveAgent({ + companyId, + name: "Research Lead", + role: "Researcher", + title: "Research Lead", + status: "active", + reportsTo: null, + connectorId: "openai-codex", + workspaceId, + model: "gpt-5.4", + capabilities: "Research and synthesis", + budgetMonthlyUsd: 180, + department: "research", + }); + const taskId = db.core.tasks.saveTask({ + companyId, + title: "Draft market report", + description: "Produce a market report with concrete evidence and business insight.", + assigneeAgentId: agentId, + workspaceId, + priority: "high", + status: "in_review", + }); + const documentId = db.core.documents.saveDocument({ + companyId, + type: "knowledge_article", + title: "Market report", + content: "Draft report body", + authorAgentId: agentId, + status: "in_review", + }); + const approvalId = db.core.approvals.requestApproval({ + companyId, + relatedTaskId: taskId, + requestedByAgentId: agentId, + type: "deliverable_review", + payloadSummary: "Review market report deliverable", + impactSummary: "Board review required before the report is finalized.", + payloadJson: JSON.stringify({ documentIds: [documentId] }), + }); + + db.core.approvals.decideApproval({ + approvalId, + state: "revision_requested", + decisionNote: "Add stronger competitor evidence and a clearer recommendation.", + }); + + const dispatchRunToWorker = vi.fn(); + const heartbeat = createHeartbeat( + { + db, + logger: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() }, + publishDomainChanged: vi.fn(), + emitEvent: vi.fn(), + notify: vi.fn(), + createRunToken: () => "test-token", + getApiPort: () => 3210, + dispatchAutomation: vi.fn(), + hasWorkerReady: () => true, + enqueueWorkerRun: vi.fn(), + isQuitting: () => false, + listRuntimeSkillEntries: () => [], + budgetService: { + autoCloseExpiredIncidents: vi.fn(() => 0), + evaluateBudgetGate: vi.fn(() => ({ allowed: true, blockedBy: [] })), + } as never, + checkpointEngine: { shouldPauseHeartbeat: () => false }, + }, + () => ({ + workspace: null, + lockMode: "none", + source: "none", + observationPath: profileDir, + }), + () => null, + dispatchRunToWorker, + (filter) => db.core.runs.hasActiveRun(filter), + ); + + handleApprovalResolved({ + db, + logger: { info: () => {}, warn: () => {}, error: () => {} }, + publishDomainChanged: () => {}, + notify: vi.fn(), + wakeAgentIfPossible: heartbeat.wakeAgentIfPossible, + queueBrowserActionExecution: () => {}, + isQuitting: () => false, + }, { + dispatchAutomation: vi.fn(), + handleTaskStatusChange: vi.fn(), + runWorkflowPipeline: async () => {}, + trackTimer: vi.fn(), + }, approvalId, companyId, "revision_requested"); + + const task = db.core.entity.getTask(taskId); + expect(task.status).toBe("in_progress"); + expect(db.core.runs.hasActiveRun({ companyId, taskId, agentId })).toBe(true); + expect(dispatchRunToWorker).toHaveBeenCalledTimes(1); + expect(dispatchRunToWorker).toHaveBeenCalledWith(expect.objectContaining({ + companyId, + task: expect.objectContaining({ id: taskId }), + trigger: "revision_needed", + })); + db.close(); }); }); diff --git a/tests/unit/connector-catalog.test.ts b/tests/unit/connector-catalog.test.ts index 1734c049..b53f4ffd 100644 --- a/tests/unit/connector-catalog.test.ts +++ b/tests/unit/connector-catalog.test.ts @@ -7,14 +7,16 @@ import { findConnectorDefinition, listFeaturedOnboardingConnectors } from "../.. * Settings panel) depends on. */ describe("CONNECTORS catalog", () => { - it("contains exactly 26 connectors (5 OAuth + 18 API key + 3 Cloud)", () => { - expect(CONNECTORS.length).toBe(26); + it("contains exactly 28 connectors (5 OAuth + 18 API key + 3 Cloud + 2 Local)", () => { + expect(CONNECTORS.length).toBe(28); const oauth = CONNECTORS.filter((c) => c.auth?.type === "oauth"); const apiKey = CONNECTORS.filter((c) => c.auth?.type === "api_key"); const cloud = CONNECTORS.filter((c) => c.auth ? ["aws", "azure", "gcp_adc"].includes(c.auth.type) : false); + const local = CONNECTORS.filter((c) => c.auth?.type === "local"); expect(oauth.length).toBe(5); expect(apiKey.length).toBe(18); expect(cloud.length).toBe(3); + expect(local.length).toBe(2); }); it("uses unique connector ids", () => { @@ -39,7 +41,7 @@ describe("CONNECTORS catalog", () => { "minimax", "minimax-cn", "huggingface", "opencode", "opencode-go", "kimi-coding", ]); // Providers wired through src/main/model-resolver.ts instead of pi's catalog. - const localProviders = new Set(["fireworks", "fireworks-firepass"]); + const localProviders = new Set(["fireworks", "fireworks-firepass", "ollama", "lm-studio"]); for (const connector of CONNECTORS) { const provider = connector.provider ?? ""; const recognised = knownPiProviders.has(provider) || localProviders.has(provider); diff --git a/tests/unit/queued-run-recovery.test.ts b/tests/unit/queued-run-recovery.test.ts new file mode 100644 index 00000000..d05062cf --- /dev/null +++ b/tests/unit/queued-run-recovery.test.ts @@ -0,0 +1,185 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { AppDatabase } from "../../src/main/database"; +import { createBudgetEnforcementService } from "../../src/main/budget-enforcement-service"; +import { createRunDispatch } from "../../src/main/orchestration/run-dispatch"; +import { createTaskRunner } from "../../src/main/orchestration/task-runner"; + +vi.mock("electron", () => ({ + safeStorage: { + isEncryptionAvailable: () => false, + getSelectedStorageBackend: () => "basic_text", + encryptString: (value: string) => Buffer.from(value, "utf8"), + decryptString: (value: Buffer) => value.toString("utf8"), + }, +})); + +const tempDirs: string[] = []; + +async function createTempDir(prefix: string) { + const dir = await mkdtemp(join(tmpdir(), prefix)); + tempDirs.push(dir); + return dir; +} + +afterEach(async () => { + await Promise.all(tempDirs.splice(0).map((dir) => rm(dir, { recursive: true, force: true }))); +}); + +function createLogger() { + return { + debug: vi.fn(), + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + }; +} + +function createHarness(db: AppDatabase) { + const logger = createLogger(); + const enqueueWorkerRun = vi.fn(); + const budgetService = createBudgetEnforcementService({ + db, + logger, + notify: vi.fn(), + publishDomainChanged: vi.fn(), + }); + const deps = { + db, + logger, + publishDomainChanged: vi.fn(), + emitEvent: vi.fn(), + notify: vi.fn(), + createRunToken: () => "token", + getApiPort: () => 3210, + dispatchAutomation: vi.fn(), + hasWorkerReady: () => true, + enqueueWorkerRun, + isQuitting: () => false, + listRuntimeSkillEntries: () => [], + budgetService, + checkpointEngine: { shouldPauseHeartbeat: () => false }, + }; + const runDispatch = createRunDispatch(deps); + const taskRunner = createTaskRunner( + deps, + runDispatch.resolveWorkspace, + runDispatch.getRunEligibilityFailure, + runDispatch.dispatchRunToWorker, + (filter) => db.core.runs.hasActiveRun(filter), + ); + return { enqueueWorkerRun, taskRunner }; +} + +function createCompanyGraph(db: AppDatabase, profileDir: string) { + db.core.connectors.updateConnectorHealth({ + id: "openai-codex", + status: "ready", + authState: "authenticated", + version: "test", + lastError: null, + }); + const companyId = db.core.company.saveCompany({ + name: "Queued Recovery Co", + description: "Recovers queued runs deterministically", + status: "active", + }); + const workspaceId = db.core.workspaces.saveWorkspace({ + companyId, + name: "Primary", + localPath: profileDir, + repoUrl: "", + repoRef: "", + isPrimary: true, + }); + const agentId = db.core.agents.saveAgent({ + companyId, + name: "Engineer", + role: "Engineer", + title: "Engineer", + status: "idle", + reportsTo: null, + connectorId: "openai-codex", + workspaceId, + model: "gpt-5.4", + capabilities: "Runs queued recovery tasks", + budgetMonthlyUsd: 100, + }); + return { companyId, workspaceId, agentId }; +} + +function createQueuedTaskRun(db: AppDatabase, input: { + companyId: string; + workspaceId: string; + agentId: string; + title: string; +}) { + const taskId = db.core.tasks.saveTask({ + companyId: input.companyId, + title: input.title, + description: "Queued before worker startup recovery.", + assigneeAgentId: input.agentId, + workspaceId: input.workspaceId, + priority: "high", + status: "todo", + }); + const run = db.core.runs.createTaskRun({ + companyId: input.companyId, + taskId, + agentId: input.agentId, + workspaceId: input.workspaceId, + connectorId: "openai-codex", + }); + if (!run) throw new Error("Expected queued run to be created."); + return { taskId, runId: run.id }; +} + +describe("queued run recovery", () => { + it("dispatches the first queued run and releases same-agent queued siblings", async () => { + const profileDir = await createTempDir("agentcompany-queued-recovery-"); + const db = new AppDatabase(profileDir); + db.init(); + const { companyId, workspaceId, agentId } = createCompanyGraph(db, profileDir); + const first = createQueuedTaskRun(db, { companyId, workspaceId, agentId, title: "First queued task" }); + const second = createQueuedTaskRun(db, { companyId, workspaceId, agentId, title: "Second queued task" }); + const { enqueueWorkerRun, taskRunner } = createHarness(db); + + taskRunner.resumeQueuedRuns(); + + expect(enqueueWorkerRun).toHaveBeenCalledOnce(); + expect(enqueueWorkerRun.mock.calls[0]?.[0]).toEqual(expect.objectContaining({ + runId: first.runId, + taskId: first.taskId, + agentId, + trigger: "recovery", + })); + expect(db.core.runs.getRun(second.runId).status).toBe("cancelled"); + expect(db.core.entity.getTask(second.taskId).status).toBe("todo"); + db.close(); + }); + + it("releases queued runs when recovery is blocked by connector readiness", async () => { + const profileDir = await createTempDir("agentcompany-queued-connector-block-"); + const db = new AppDatabase(profileDir); + db.init(); + const { companyId, workspaceId, agentId } = createCompanyGraph(db, profileDir); + const queued = createQueuedTaskRun(db, { companyId, workspaceId, agentId, title: "Connector blocked task" }); + db.core.connectors.updateConnectorHealth({ + id: "openai-codex", + status: "auth_required", + authState: "missing", + version: "test", + lastError: "Login required", + }); + const { enqueueWorkerRun, taskRunner } = createHarness(db); + + taskRunner.resumeQueuedRuns(); + + expect(enqueueWorkerRun).not.toHaveBeenCalled(); + expect(db.core.runs.getRun(queued.runId).status).toBe("cancelled"); + expect(db.core.entity.getTask(queued.taskId).status).toBe("todo"); + db.close(); + }); +}); diff --git a/tests/unit/scheduler-heartbeat.test.ts b/tests/unit/scheduler-heartbeat.test.ts index 1b56ca40..82d91ecf 100644 --- a/tests/unit/scheduler-heartbeat.test.ts +++ b/tests/unit/scheduler-heartbeat.test.ts @@ -199,6 +199,38 @@ describe("unified scheduler heartbeat cadence", () => { expect(wakeAgentIfPossible).toHaveBeenCalledWith(agentId, companyId, "timer"); }); + it("does not wake assigned work for an agent with heartbeat disabled", () => { + const companyId = createCompany(db); + createConnector(db); + const workspaceId = createWorkspace(db, companyId, tempDir); + const agentId = createAgent(db, companyId, workspaceId); + db.core.agents.setHeartbeat(agentId, false, 1); + db.core.tasks.saveTask({ + companyId, + title: "Paused operating work", + description: "Manual heartbeat disable should pause periodic recovery.", + assigneeAgentId: agentId, + workspaceId, + priority: "high", + status: "todo", + }); + const wakeAgentIfPossible = vi.fn(); + const scheduler = createScheduler( + createDeps(db), + () => false, + wakeAgentIfPossible, + ); + + try { + scheduler.startAllSchedulers(); + vi.advanceTimersByTime(60_000); + } finally { + scheduler.stopAllSchedulers(); + } + + expect(wakeAgentIfPossible).not.toHaveBeenCalled(); + }); + it("wakes an agent with a queued wakeup request on heartbeat cadence", () => { const companyId = createCompany(db); createConnector(db); diff --git a/tests/unit/worker-completion-boundary.test.ts b/tests/unit/worker-completion-boundary.test.ts index c68da7ee..d72cda6d 100644 --- a/tests/unit/worker-completion-boundary.test.ts +++ b/tests/unit/worker-completion-boundary.test.ts @@ -24,19 +24,45 @@ function runFinishedMessage(runId: string) { }; } -function createHandlerDeps(calls: string[]) { - const releaseWorkspaceLock = vi.fn(() => calls.push("releaseWorkspaceLock")); +function createHandlerDeps(calls: string[], options: { + taskId?: string | null; + taskStatus?: string; + sessionAccumulateThrows?: boolean; + processRunCompletionThrows?: boolean; + activityThrows?: boolean; + emitEventThrows?: boolean; + wakeupThrows?: boolean; + releaseWorkspaceThrows?: boolean; + markRunCompletionFailureThrows?: boolean; +} = {}) { + const releaseWorkspaceLock = vi.fn(() => { + calls.push("releaseWorkspaceLock"); + if (options.releaseWorkspaceThrows) { + throw new Error("workspace release failed"); + } + }); const finishRun = vi.fn(() => { calls.push("finishRun"); return { budgetHardStopped: false }; }); const rotateSession = vi.fn(() => calls.push("rotateSession")); - const processRunCompletion = vi.fn(() => calls.push("processRunCompletion")); + const processRunCompletion = vi.fn(() => { + calls.push("processRunCompletion"); + if (options.processRunCompletionThrows) { + throw new Error("processRunCompletion failed"); + } + }); const releaseRunResources = vi.fn((): RunResourceReleaseResult => { calls.push("releaseRunResources"); return { ok: true as const }; }); - const markRunCompletionFailure = vi.fn(() => calls.push("markRunCompletionFailure")); + const markRunCompletionFailure = vi.fn(() => { + calls.push("markRunCompletionFailure"); + if (options.markRunCompletionFailureThrows) { + throw new Error("mark completion failed"); + } + }); + const advanceTaskStatus = vi.fn(() => calls.push("advanceTaskStatus")); const db = { core: { @@ -45,14 +71,21 @@ function createHandlerDeps(calls: string[]) { id: RUN_ID, companyId: "company-1", agentId: "agent-1", - taskId: null, + taskId: options.taskId ?? null, workspaceId: "workspace-1", })), finishRun, markRunCompletionFailure, }, workspaces: { releaseWorkspaceLock }, - activity: { addActivity: vi.fn(() => calls.push("addActivity")) }, + activity: { + addActivity: vi.fn(() => { + calls.push("addActivity"); + if (options.activityThrows) { + throw new Error("activity write failed"); + } + }), + }, entity: { getAgent: vi.fn(() => ({ id: "agent-1", @@ -60,11 +93,19 @@ function createHandlerDeps(calls: string[]) { reportsTo: "manager-1", sessionId: "session-1", })), + getTask: vi.fn(() => ({ + id: options.taskId ?? "task-1", + companyId: "company-1", + agentId: "agent-1", + title: "Completion task", + status: options.taskStatus ?? "in_progress", + })), countActiveGoalsForCompany: vi.fn(() => 0), }, tasks: { getSessionById: vi.fn(() => ({ id: "session-1" })), listAgentTasks: vi.fn(() => []), + advanceTaskStatus, }, agents: { getDirectReports: vi.fn(() => []), @@ -76,14 +117,24 @@ function createHandlerDeps(calls: string[]) { }, automation: { wakeup: { - promoteWakeupRequests: vi.fn(() => 0), + promoteWakeupRequests: vi.fn(() => { + if (options.wakeupThrows) { + throw new Error("wakeup promotion failed"); + } + return 0; + }), getPreferredWakeTrigger: vi.fn(() => null), }, }, }; const sessionService = { - accumulateTokenUsage: vi.fn(() => calls.push("accumulateTokenUsage")), + accumulateTokenUsage: vi.fn(() => { + calls.push("accumulateTokenUsage"); + if (options.sessionAccumulateThrows) { + throw new Error("session accumulation failed"); + } + }), evaluateRotation: vi.fn(() => ({ shouldRotate: true, reason: "unit-test" })), rotateSession, }; @@ -98,7 +149,12 @@ function createHandlerDeps(calls: string[]) { error: vi.fn(), }, isQuitting: vi.fn(() => false), - emitEvent: vi.fn(() => calls.push("emitEvent")), + emitEvent: vi.fn(() => { + calls.push("emitEvent"); + if (options.emitEventThrows) { + throw new Error("emit failed"); + } + }), wakeAgentIfPossible: vi.fn(() => null), processRunCompletion, notify: vi.fn(), @@ -111,6 +167,7 @@ function createHandlerDeps(calls: string[]) { processRunCompletion, releaseRunResources, markRunCompletionFailure, + advanceTaskStatus, }; } @@ -227,6 +284,265 @@ describe("worker completion boundary", () => { expect(worker.postMessage).toHaveBeenCalledWith({ type: "run-completion-ack", runId: RUN_ID }); }); + it("continues completion when artifact failure marking throws", async () => { + const calls: string[] = []; + const { + deps, + releaseWorkspaceLock, + processRunCompletion, + releaseRunResources, + markRunCompletionFailure, + } = createHandlerDeps(calls, { markRunCompletionFailureThrows: true }); + releaseRunResources.mockImplementationOnce(() => { + calls.push("releaseRunResources"); + return { ok: false as const, errorMessage: "completion_reconcile failed" }; + }); + const worker = { postMessage: vi.fn(() => calls.push("completionAck")) }; + + await handleWorkerMessage( + runFinishedMessage(RUN_ID), + worker as never, + deps as never, + workerMessageSchema, + false, + 2, + 30_000, + new Map(), + 3, + new Set(), + new Map(), + ); + + expect(markRunCompletionFailure).toHaveBeenCalledWith(RUN_ID, "completion_reconcile failed"); + expect(processRunCompletion).toHaveBeenCalledWith( + expect.objectContaining({ id: RUN_ID }), + expect.objectContaining({ + status: "failed", + errorMessage: "completion_reconcile failed", + usageStatus: "error_stop", + }), + ); + expect(releaseWorkspaceLock).toHaveBeenCalledWith("workspace-1", RUN_ID); + expect(worker.postMessage).toHaveBeenCalledWith({ type: "run-completion-ack", runId: RUN_ID }); + }); + + it("continues completion automation when session accumulation fails", async () => { + const calls: string[] = []; + const { + deps, + releaseWorkspaceLock, + processRunCompletion, + } = createHandlerDeps(calls, { sessionAccumulateThrows: true }); + const worker = { postMessage: vi.fn(() => calls.push("completionAck")) }; + + await handleWorkerMessage( + runFinishedMessage(RUN_ID), + worker as never, + deps as never, + workerMessageSchema, + false, + 2, + 30_000, + new Map(), + 3, + new Set(), + new Map(), + ); + + expect(processRunCompletion).toHaveBeenCalled(); + expect(releaseWorkspaceLock).toHaveBeenCalledWith("workspace-1", RUN_ID); + expect(worker.postMessage).toHaveBeenCalledWith({ type: "run-completion-ack", runId: RUN_ID }); + }); + + it("marks completion failed and blocks the task when completion automation throws", async () => { + const calls: string[] = []; + const { + deps, + releaseWorkspaceLock, + markRunCompletionFailure, + advanceTaskStatus, + } = createHandlerDeps(calls, { + taskId: "task-1", + taskStatus: "in_progress", + processRunCompletionThrows: true, + }); + const worker = { postMessage: vi.fn(() => calls.push("completionAck")) }; + + await handleWorkerMessage( + runFinishedMessage(RUN_ID), + worker as never, + deps as never, + workerMessageSchema, + false, + 2, + 30_000, + new Map(), + 3, + new Set(), + new Map(), + ); + + expect(markRunCompletionFailure).toHaveBeenCalledWith(RUN_ID, "processRunCompletion failed"); + expect(advanceTaskStatus).toHaveBeenCalledWith( + "task-1", + "blocked", + expect.stringContaining("Run completion automation failed: processRunCompletion failed"), + ); + expect(releaseWorkspaceLock).toHaveBeenCalledWith("workspace-1", RUN_ID); + expect(worker.postMessage).toHaveBeenCalledWith({ type: "run-completion-ack", runId: RUN_ID }); + }); + + it("releases workspace and acknowledges completion when activity logging fails", async () => { + const calls: string[] = []; + const { + deps, + releaseWorkspaceLock, + processRunCompletion, + } = createHandlerDeps(calls, { activityThrows: true }); + const worker = { postMessage: vi.fn(() => calls.push("completionAck")) }; + + await handleWorkerMessage( + runFinishedMessage(RUN_ID), + worker as never, + deps as never, + workerMessageSchema, + false, + 2, + 30_000, + new Map(), + 3, + new Set(), + new Map(), + ); + + expect(processRunCompletion).toHaveBeenCalled(); + expect(releaseWorkspaceLock).toHaveBeenCalledWith("workspace-1", RUN_ID); + expect(worker.postMessage).toHaveBeenCalledWith({ type: "run-completion-ack", runId: RUN_ID }); + }); + + it("releases workspace and acknowledges completion when renderer event emission fails", async () => { + const calls: string[] = []; + const { + deps, + releaseWorkspaceLock, + processRunCompletion, + } = createHandlerDeps(calls, { emitEventThrows: true }); + const worker = { postMessage: vi.fn(() => calls.push("completionAck")) }; + + await handleWorkerMessage( + runFinishedMessage(RUN_ID), + worker as never, + deps as never, + workerMessageSchema, + false, + 2, + 30_000, + new Map(), + 3, + new Set(), + new Map(), + ); + + expect(processRunCompletion).toHaveBeenCalled(); + expect(releaseWorkspaceLock).toHaveBeenCalledWith("workspace-1", RUN_ID); + expect(worker.postMessage).toHaveBeenCalledWith({ type: "run-completion-ack", runId: RUN_ID }); + }); + + it("releases workspace and acknowledges completion when deferred wake promotion fails", async () => { + const calls: string[] = []; + const { + deps, + releaseWorkspaceLock, + processRunCompletion, + } = createHandlerDeps(calls, { wakeupThrows: true }); + const worker = { postMessage: vi.fn(() => calls.push("completionAck")) }; + + await handleWorkerMessage( + runFinishedMessage(RUN_ID), + worker as never, + deps as never, + workerMessageSchema, + false, + 2, + 30_000, + new Map(), + 3, + new Set(), + new Map(), + ); + + expect(processRunCompletion).toHaveBeenCalled(); + expect(releaseWorkspaceLock).toHaveBeenCalledWith("workspace-1", RUN_ID); + expect(worker.postMessage).toHaveBeenCalledWith({ type: "run-completion-ack", runId: RUN_ID }); + }); + + it("acknowledges completion when workspace release fails", async () => { + const calls: string[] = []; + const { + deps, + releaseWorkspaceLock, + processRunCompletion, + } = createHandlerDeps(calls, { releaseWorkspaceThrows: true }); + const worker = { postMessage: vi.fn(() => calls.push("completionAck")) }; + + await handleWorkerMessage( + runFinishedMessage(RUN_ID), + worker as never, + deps as never, + workerMessageSchema, + false, + 2, + 30_000, + new Map(), + 3, + new Set(), + new Map(), + ); + + expect(processRunCompletion).toHaveBeenCalled(); + expect(releaseWorkspaceLock).toHaveBeenCalledWith("workspace-1", RUN_ID); + expect(worker.postMessage).toHaveBeenCalledWith({ type: "run-completion-ack", runId: RUN_ID }); + }); + + it("blocks the task and acknowledges completion when marking completion failure throws", async () => { + const calls: string[] = []; + const { + deps, + releaseWorkspaceLock, + markRunCompletionFailure, + advanceTaskStatus, + } = createHandlerDeps(calls, { + taskId: "task-1", + taskStatus: "in_progress", + processRunCompletionThrows: true, + markRunCompletionFailureThrows: true, + }); + const worker = { postMessage: vi.fn(() => calls.push("completionAck")) }; + + await handleWorkerMessage( + runFinishedMessage(RUN_ID), + worker as never, + deps as never, + workerMessageSchema, + false, + 2, + 30_000, + new Map(), + 3, + new Set(), + new Map(), + ); + + expect(markRunCompletionFailure).toHaveBeenCalledWith(RUN_ID, "processRunCompletion failed"); + expect(advanceTaskStatus).toHaveBeenCalledWith( + "task-1", + "blocked", + expect.stringContaining("Run completion automation failed: processRunCompletion failed"), + ); + expect(releaseWorkspaceLock).toHaveBeenCalledWith("workspace-1", RUN_ID); + expect(worker.postMessage).toHaveBeenCalledWith({ type: "run-completion-ack", runId: RUN_ID }); + }); + it("keeps same-workspace runs queued until parent acknowledges completion", async () => { const parentPort = new EventEmitter() as EventEmitter & { postMessage: ReturnType; diff --git a/tests/unit/worker-crash-recovery.test.ts b/tests/unit/worker-crash-recovery.test.ts new file mode 100644 index 00000000..94771f86 --- /dev/null +++ b/tests/unit/worker-crash-recovery.test.ts @@ -0,0 +1,201 @@ +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import { mkdtemp, rm } from "node:fs/promises"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +vi.mock("electron", () => ({ + safeStorage: { + isEncryptionAvailable: () => false, + getSelectedStorageBackend: () => "basic_text", + encryptString: (value: string) => Buffer.from(value, "utf8"), + decryptString: (value: Buffer) => value.toString("utf8"), + }, +})); + +import { AppDatabase } from "../../src/main/database"; +import { handleWorkerExit, requeueQueuedRunsAfterWorkerRestart } from "../../src/main/worker/entry/crash-recovery"; + +function createLogger() { + return { + debug: vi.fn(), + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + }; +} + +function createCompany(db: AppDatabase): string { + return db.core.company.saveCompany({ + name: "Crash Recovery Co", + description: "Worker crash recovery contract test", + status: "active", + }); +} + +function createConnector(db: AppDatabase): void { + db.core.connectors.saveConnector({ + id: "openai-codex", + model: "gpt-5.4", + envBindingText: "", + notes: "Test connector", + }); +} + +function createWorkspace(db: AppDatabase, companyId: string, localPath: string): string { + return db.core.workspaces.saveWorkspace({ + companyId, + name: "Primary", + localPath, + repoUrl: "", + repoRef: "", + isPrimary: true, + }); +} + +function createAgent(db: AppDatabase, companyId: string, workspaceId: string): string { + return db.core.agents.saveAgent({ + companyId, + name: "Recovery Agent", + role: "Engineer", + title: "Engineer", + status: "active", + reportsTo: null, + connectorId: "openai-codex", + workspaceId, + model: "gpt-5.4", + capabilities: "Continues work after runtime crashes", + budgetMonthlyUsd: 100, + }); +} + +describe("worker crash recovery", () => { + let tempDir: string; + let db: AppDatabase; + + beforeEach(async () => { + tempDir = await mkdtemp(join(tmpdir(), "agentcompany-worker-crash-")); + db = new AppDatabase(tempDir); + db.init(); + }); + + afterEach(async () => { + db.close(); + await rm(tempDir, { recursive: true, force: true }); + }); + + it("returns interrupted running tasks to checkoutable work and wakes the original agent", () => { + const companyId = createCompany(db); + createConnector(db); + const workspaceId = createWorkspace(db, companyId, tempDir); + const agentId = createAgent(db, companyId, workspaceId); + const taskId = db.core.tasks.saveTask({ + companyId, + title: "Continue Phase 2", + description: "This task must remain recoverable after the worker process exits.", + assigneeAgentId: agentId, + workspaceId, + priority: "high", + status: "todo", + }); + const run = db.core.runs.createTaskRun({ + companyId, + taskId, + agentId, + workspaceId, + connectorId: "openai-codex", + }); + expect(run).not.toBeNull(); + db.core.runs.markRunStatus(run!.id, "running", "Started"); + expect(db.core.entity.getTask(taskId).status).toBe("in_progress"); + expect(db.core.runs.hasActiveRun({ companyId, agentId })).toBe(true); + + const wakeAgentIfPossible = vi.fn(() => null); + handleWorkerExit( + 1, + { + db, + logger: createLogger(), + isQuitting: () => false, + publishDomainChanged: vi.fn(), + emitEvent: vi.fn(), + wakeAgentIfPossible, + }, + ); + + expect(db.core.runs.getRun(run!.id).status).toBe("interrupted"); + expect(db.core.entity.getTask(taskId).status).toBe("todo"); + expect(db.core.runs.hasActiveRun({ companyId, agentId })).toBe(false); + expect(wakeAgentIfPossible).toHaveBeenCalledWith(agentId, companyId, "recovery"); + + const retryRun = db.core.runs.createTaskRun({ + companyId, + taskId, + agentId, + workspaceId, + connectorId: "openai-codex", + }); + expect(retryRun).not.toBeNull(); + }); + + it("requeues queued runs after the replacement worker is ready", () => { + const companyId = createCompany(db); + createConnector(db); + const workspaceId = createWorkspace(db, companyId, tempDir); + const agentId = createAgent(db, companyId, workspaceId); + const taskId = db.core.tasks.saveTask({ + companyId, + title: "Queued recovery task", + description: "This run was queued when the worker process exited.", + assigneeAgentId: agentId, + workspaceId, + priority: "high", + status: "todo", + }); + const run = db.core.runs.createTaskRun({ + companyId, + taskId, + agentId, + workspaceId, + connectorId: "openai-codex", + }); + expect(run).not.toBeNull(); + expect(db.core.runs.getRun(run!.id).status).toBe("queued"); + expect(db.core.runs.hasActiveRun({ companyId, agentId })).toBe(true); + + const dispatchRunToWorker = vi.fn(); + const requeueCandidates = handleWorkerExit( + 1, + { + db, + logger: createLogger(), + isQuitting: () => false, + publishDomainChanged: vi.fn(), + emitEvent: vi.fn(), + wakeAgentIfPossible: vi.fn(() => null), + }, + ); + + expect(dispatchRunToWorker).not.toHaveBeenCalled(); + expect(requeueCandidates).toHaveLength(1); + expect(db.core.runs.getRun(run!.id).status).toBe("queued"); + + requeueQueuedRunsAfterWorkerRestart( + requeueCandidates, + { + db, + logger: createLogger(), + dispatchRunToWorker, + }, + ); + + expect(dispatchRunToWorker).toHaveBeenCalledOnce(); + expect(dispatchRunToWorker).toHaveBeenCalledWith(expect.objectContaining({ + runId: run!.id, + companyId, + task: expect.objectContaining({ id: taskId }), + agent: expect.objectContaining({ id: agentId }), + trigger: "recovery", + })); + expect(db.core.runs.hasActiveRun({ companyId, agentId })).toBe(true); + }); +}); diff --git a/tests/unit/worker-service-shutdown.test.ts b/tests/unit/worker-service-shutdown.test.ts index 72698174..909828a4 100644 --- a/tests/unit/worker-service-shutdown.test.ts +++ b/tests/unit/worker-service-shutdown.test.ts @@ -1,5 +1,5 @@ import { EventEmitter } from "node:events"; -import { describe, expect, it, vi } from "vitest"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; const mocks = vi.hoisted(() => ({ forkMock: vi.fn(), @@ -25,6 +25,15 @@ function createWorker() { } describe("worker service shutdown", () => { + beforeEach(() => { + mocks.forkMock.mockReset(); + }); + + afterEach(() => { + vi.clearAllTimers(); + vi.useRealTimers(); + }); + it("keeps completed run outcomes when cleaning shutdown runs", async () => { const worker = createWorker(); mocks.forkMock.mockReturnValueOnce(worker); @@ -83,4 +92,128 @@ describe("worker service shutdown", () => { status: "interrupted", })); }); + + it("clears the crashed worker reference before recovery runs", () => { + vi.useFakeTimers(); + const worker = createWorker(); + mocks.forkMock.mockReturnValueOnce(worker); + const publishDomainChanged = vi.fn(); + const service = createWorkerService({ + db: { + projection: { + snapshot: { + listSnapshot: vi.fn(() => ({ runs: [] })), + }, + }, + }, + logger: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() }, + workerPath: "/tmp/worker.js", + isQuitting: () => false, + publishDomainChanged, + emitEvent: vi.fn(), + wakeAgentIfPossible: vi.fn(() => null), + dispatchRunToWorker: vi.fn(), + processRunCompletion: vi.fn(), + onRunStarted: vi.fn(), + onRunCompleted: vi.fn(), + notify: vi.fn(), + budgetService: undefined, + sessionService: undefined, + getRuntimeBridge: vi.fn(() => null), + getRuntimeMessageInjector: vi.fn(() => null), + } as never); + + service.createWorker(); + expect(service.getWorker()).toBe(worker); + worker.emit("exit", 1); + + expect(service.getWorker()).toBeNull(); + expect(publishDomainChanged).toHaveBeenCalledOnce(); + }); + + it("requeues queued crash runs after the replacement worker is created", () => { + vi.useFakeTimers(); + const crashedWorker = createWorker(); + const replacementWorker = createWorker(); + mocks.forkMock + .mockReturnValueOnce(crashedWorker) + .mockReturnValueOnce(replacementWorker); + const queuedRun = { + id: "run-queued", + status: "queued", + taskId: "task-1", + agentId: "agent-1", + companyId: "company-1", + workspaceId: null, + connectorId: "openai-codex", + }; + const dispatchRunToWorker = vi.fn(); + const service = createWorkerService({ + db: { + projection: { + snapshot: { + listSnapshot: vi.fn(() => ({ + runs: [queuedRun], + companies: [{ id: "company-1", name: "Company", description: "Recovered company" }], + })), + }, + }, + core: { + runs: { + getRun: vi.fn(() => queuedRun), + }, + entity: { + getTask: vi.fn(() => ({ + id: "task-1", + companyId: "company-1", + workspaceId: null, + projectId: null, + requiresFilesystem: false, + })), + getAgent: vi.fn(() => ({ + id: "agent-1", + companyId: "company-1", + connectorId: "openai-codex", + workspaceId: null, + })), + }, + connectors: { + getConnector: vi.fn(() => ({ id: "openai-codex" })), + }, + workspaces: { + releaseWorkspaceLock: vi.fn(), + getCompanyObservationWorkspace: vi.fn(() => null), + }, + }, + }, + logger: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() }, + workerPath: "/tmp/worker.js", + isQuitting: () => false, + publishDomainChanged: vi.fn(), + emitEvent: vi.fn(), + wakeAgentIfPossible: vi.fn(() => null), + dispatchRunToWorker, + processRunCompletion: vi.fn(), + onRunStarted: vi.fn(), + onRunCompleted: vi.fn(), + notify: vi.fn(), + budgetService: undefined, + sessionService: undefined, + getRuntimeBridge: vi.fn(() => null), + getRuntimeMessageInjector: vi.fn(() => null), + } as never); + + service.createWorker(); + crashedWorker.emit("exit", 1); + expect(dispatchRunToWorker).not.toHaveBeenCalled(); + + vi.advanceTimersByTime(1_000); + + expect(service.getWorker()).toBe(replacementWorker); + expect(dispatchRunToWorker).toHaveBeenCalledWith(expect.objectContaining({ + runId: "run-queued", + companyId: "company-1", + trigger: "recovery", + })); + }); });