maestro/src/worker.ts
oss-sync d41ff0f658
Some checks failed
CI / build-and-test (push) Failing after 6m23s
sync: update from private repo (d3780b00)
2026-07-09 00:12:24 +00:00

3211 lines
143 KiB
TypeScript
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

import { Repository, Job, localTaskRepoName, type JobRole } from './db/repository.js';
import { userPiecesDir } from './user-folder/paths.js';
import { BrowserSessionRepo } from './db/browser-session-repo.js';
import { assertProfileOwner } from './engine/browser-session-auth.js';
import { decryptProfileState } from './crypto/profile-dek.js';
import { OpenAICompatClient } from './llm/openai-compat.js';
import { llmRoutingKey, shouldRequeueForModelMismatch } from './llm/routing-key.js';
import { loadPiece, runPiece, PieceRunCallbacks, PieceDef, type PieceRunResult } from './engine/piece-runner.js';
import { LocalProgressReporter } from './progress/local-reporter.js';
import { buildLocalConversationContext } from './engine/local-context.js';
import { AppConfig, isExecutionWorker, resolveMaxStreamMs, type WorkerDef, type ReflectionConfig } from './config.js';
import { existsSync, mkdirSync, readdirSync, readFileSync, statSync, writeFileSync } from 'fs';
import { join } from 'path';
import { logger } from './logger.js';
import { commitWorkspaceChanges, ensureWorkspaceGitRepo } from './git/workspace-manager.js';
import { spaceConflictDir, resolveReporterRuntimeDir } from './spaces/runtime-paths.js';
import { ContextManager, fetchOllamaContextLimit } from './engine/context-manager.js';
import { Conversation } from './engine/context/conversation.js';
import { summarizeToolInput, type ActivityLogMetadata } from './progress/log-format.js';
import { ensureKeepaGraphs } from './engine/tools/amazon.js';
import type { McpTokenManager } from './mcp/token-manager.js';
import { mergeMcpConfig } from './mcp/config.js';
import { createStickyBackendResolver } from './worker/sticky-backend.js';
import { mapPieceStatusToMetric, type JobMetricStatus } from './worker/piece-metric-status.js';
import { pickIdlerIndex } from './worker/idle-routing.js';
import { jobEventBus } from './bridge/job-events.js';
import { normalizeToolNameForMetric } from './metrics/tool-name-allowlist.js';
import { parseToolPolicy, resolveWorkspaceTools } from './engine/workspace-tool-policy.js';
import { resolvePythonPackagesConfig, spaceKeyFor, spaceCurrentDir } from './engine/python-packages.js';
import { getSshSubsystem } from './engine/tools/ssh.js';
import { DEADLINE_ABORT_REASON, ABORT_CODE_CANCELLED, ABORT_CODE_DEADLINE, raceWithGrace } from './engine/cancellation.js';
const RETRY_HANDOFF_MAX_LENGTH = 8_000;
const RETRY_DIAGNOSTICS_PREVIEW_LENGTH = 1_200;
const RETRY_LESSONS_MAX_LINES = 12;
/**
* Default hard ceiling (minutes) for a single job's active execution. A job
* that runs longer than this — for any reason, including a runaway LLM stream
* the agent loop can't escape — is force-aborted so its in-memory concurrency
* slot is released without a process restart. Override via safety.maxJobMinutes
* (0 disables). Generous by design: a legitimate job that yields to subtasks
* returns and frees its slot well before this fires.
*/
const DEFAULT_MAX_JOB_MINUTES = 180;
/**
* Grace window (seconds) after a deadline/cancel abort fires before the worker
* hard-kills the job. ②A wires the abort signal into the realistic blocking
* tools (web/MCP/browser) so a run normally unwinds cooperatively well within
* this window. This is the ②B safety net: if some *other* await never observes
* the abort, the worker force-marks the job terminal and releases its inflight
* slot after the grace elapses. Override via safety.deadlineGraceSeconds; a
* value of 0 DISABLES the hard-kill fallback entirely (the worker awaits the run
* directly and relies solely on ②A's cooperative abort). The still-running
* in-process work becomes a detached zombie (accepted trade-off — the slot
* matters more).
*/
const DEFAULT_DEADLINE_GRACE_SECONDS = 15;
/**
* How often (ms) the per-job guard timer polls for a cancel request and checks
* the hard deadline. cancelCheck() only runs between agent-loop iterations, so
* a cancel pressed DURING a long/stuck LLM stream would otherwise never fire
* the abort. This timer closes that gap (cancel becomes effective within one
* interval) and enforces the maxJobMinutes deadline.
*/
const JOB_GUARD_POLL_MS = 3_000;
/**
* Pure predicate for the per-job hard deadline. Force-abort when a deadline is
* configured (maxJobMs > 0), the job hasn't already been aborted, and it has
* run past the deadline. Kept pure so the guard's timing logic is testable
* without spinning up a real worker/DB.
*/
export function shouldDeadlineAbort(elapsedMs: number, maxJobMs: number, alreadyAborted: boolean): boolean {
return maxJobMs > 0 && !alreadyAborted && elapsedMs > maxJobMs;
}
/**
* Zombie guard for SpawnSubTask. After a deadline/user abort — especially a ②B
* hard-kill that already cancelled existing children and marked the job terminal
* — a detached runPiece can still reach SpawnSubTask. A new child enqueued then
* is an orphan no live parent will consume, so refuse it. Allow the spawn only
* while the run is still live: not aborted AND the DB row is still 'running'.
*/
export function canSpawnSubtask(signalAborted: boolean, dbJobStatus: string): boolean {
return !signalAborted && dbJobStatus === 'running';
}
/**
* Abort reasons that are agent-driven AND deterministic: re-running the whole
* job produces the same end state, so a retry only creates "retry" churn (a
* full, expensive re-execution that lands the job in the `retry` status). These
* are terminal on the first occurrence — never retried.
*
* `agent_self_abort` is here because the agent deliberately called
* `complete({status:'aborted'})`; retrying would ignore its decision.
*/
const HARD_TERMINAL_ABORT_REASONS: ReadonlySet<string> = new Set([
'text_only_limit',
'loop_detected',
'tool_loop_detected',
'movement_not_found',
'movement_without_transition',
'ask_limit_reached',
'max_movements_exceeded',
'movement_abort',
'agent_self_abort',
]);
/**
* Abort reasons that are non-deterministic enough that ONE re-run sometimes
* recovers (production data: max_iterations / context_overflow recovered at
* ~avg 2.1 attempts). We allow a single recovery retry, but if the very next
* attempt aborts with the SAME reason we give up — that's a stuck loop, not a
* transient hiccup. This kills the churn while preserving the occasional win.
*
* Everything else (llm_error / null / connection-fatal) stays fully retryable
* up to maxAttempts as before — those are transient infrastructure failures.
*/
const RECOVERABLE_ONCE_ABORT_REASONS: ReadonlySet<string> = new Set([
'max_iterations_exceeded',
'context_overflow',
]);
function buildTimeContextBlock(): string {
const now = new Date();
const utc = now.toISOString();
const jst = new Intl.DateTimeFormat('ja-JP', {
timeZone: 'Asia/Tokyo',
year: 'numeric',
month: '2-digit',
day: '2-digit',
hour: '2-digit',
minute: '2-digit',
second: '2-digit',
hour12: false,
weekday: 'short',
}).format(now);
return [
'## 実行時刻コンテキスト',
`- Current time (UTC): ${utc}`,
`- Current time (JST): ${jst}`,
'- 時刻依存の判断(今日/昨日/直近◯時間/最新ニュース等)は必ずこの時刻を基準に行うこと。',
'',
].join('\n');
}
function getLocalTaskId(repoName: string): number | null {
const match = /^local\/task-(\d+)$/.exec(repoName);
if (!match) return null;
return Number(match[1]);
}
function getSubTaskParentJobId(repoName: string): string | null {
const match = /^subtask\/([0-9a-f-]{36})$/.exec(repoName);
if (!match) return null;
return match[1]!;
}
/**
* Browser session を keying するための「論理タスク ID」を解決する。
*
* - local task の直接実行 → そのタスクの ID (string)
* - subtask 実行 → 親方向に最大 5 段まで walk up し、最初に見つかる
* local task の ID。subtaskDepth は config 上 2 が上限なので余裕を見て 5
* - 親が gitea issue 等 local task でないジョブの場合 → null (BrowseWeb は
* noVNC モードを使えない / 旧来の頭出しに fallback する)
*/
async function resolveSessionTaskId(
repo: Repository,
job: Job,
): Promise<{ taskId: string | undefined; userId: string | undefined }> {
const directLocalTaskId = getLocalTaskId(job.repo);
if (directLocalTaskId !== null) {
return { taskId: String(directLocalTaskId), userId: job.ownerId ?? undefined };
}
let cursor: string | null = getSubTaskParentJobId(job.repo);
let hops = 0;
while (cursor && hops < 5) {
const parent = await repo.getJob(cursor);
if (!parent) return { taskId: undefined, userId: job.ownerId ?? undefined };
const parentLocalTaskId = getLocalTaskId(parent.repo);
if (parentLocalTaskId !== null) {
return {
taskId: String(parentLocalTaskId),
// owner は親と同じはずだが、念のため fallback も用意
userId: parent.ownerId ?? job.ownerId ?? undefined,
};
}
cursor = parent.parentJobId ?? getSubTaskParentJobId(parent.repo);
hops++;
}
return { taskId: undefined, userId: job.ownerId ?? undefined };
}
/**
* Walk up the job ancestry to find the root local-task job id.
*
* - If `job` is itself a local/task-N job, return `job.id` immediately.
* - Otherwise follow parentJobId (or getSubTaskParentJobId from repo name)
* up to 5 hops, returning the id of the first ancestor whose repo matches
* local/task-N. Falls back to the last ancestor id seen if no local task
* is found within the limit.
* - Returns undefined if no parent chain exists at all.
*/
export async function resolveRootJobId(repo: Repository, job: Job): Promise<string | undefined> {
if (getLocalTaskId(job.repo) !== null) return job.id;
let cursor: string | null = job.parentJobId ?? getSubTaskParentJobId(job.repo);
let last: string | undefined = cursor ?? undefined;
let hops = 0;
while (cursor && hops < 5) {
const parent = await repo.getJob(cursor);
if (!parent) return last;
last = parent.id;
if (getLocalTaskId(parent.repo) !== null) return parent.id;
cursor = parent.parentJobId ?? getSubTaskParentJobId(parent.repo);
hops++;
}
return last;
}
/**
* Fields a derived job must inherit from the job it descends from: ownership,
* visibility, space, and — critically — the bound browser session profile.
*
* Used by every `createJob` that produces a child/continuation of an existing
* job (subtask spawn, subtask ASK re-enqueue). Centralised so the inheritance
* contract lives in ONE place: omitting `browserSessionProfileId` here makes
* BrowseWeb inside the derived job run without the parent's saved login session
* (the cookies/storageState are only loaded when the job row carries the
* profile id). Regression fixed 2026-06-26 — research-via-delegate made nearly
* all browsing happen inside subtasks, surfacing the missing inheritance.
*/
export function inheritJobContext(job: Job): {
ownerId: string | null;
visibility: 'private' | 'org' | 'public';
visibilityScopeOrgId: string | null;
spaceId: string | null;
browserSessionProfileId: number | null;
} {
return {
ownerId: job.ownerId,
visibility: job.visibility,
visibilityScopeOrgId: job.visibilityScopeOrgId,
spaceId: job.spaceId ?? null,
browserSessionProfileId: job.browserSessionProfileId ?? null,
};
}
function buildUiMetadataBlock(job: Job): string {
return [
'---',
`ui_profile: ${job.requiredRole}`,
`ui_output_format: ${/ui_output_format:\s*(text|markdown|json)/i.exec(job.instruction)?.[1]?.toLowerCase() ?? 'markdown'}`,
`ui_ask_policy: ${/ui_ask_policy:\s*(low|high)/i.exec(job.instruction)?.[1]?.toLowerCase() ?? 'low'}`,
`ui_priority: ${/ui_priority:\s*(low|medium|high)/i.exec(job.instruction)?.[1]?.toLowerCase() ?? 'medium'}`,
'---',
].join('\n');
}
function truncateRetryText(text: string, maxLength: number): string {
const trimmed = text.trim();
if (trimmed.length <= maxLength) return trimmed;
return `${trimmed.slice(0, maxLength)}...`;
}
function readRetryLessons(workspacePath: string): string[] {
const logPath = join(workspacePath, 'logs', 'lessons.jsonl');
if (!existsSync(logPath)) return [];
try {
return readFileSync(logPath, 'utf-8')
.split('\n')
.filter(Boolean)
.slice(-RETRY_LESSONS_MAX_LINES)
.map((line) => {
try {
const data = JSON.parse(line) as { movement?: string; lessons?: string };
const movement = data.movement ? `[${data.movement}] ` : '';
return `- ${movement}${truncateRetryText(String(data.lessons ?? ''), 500)}`;
} catch {
return `- ${truncateRetryText(line, 500)}`;
}
})
.filter((line) => line.trim() !== '-');
} catch {
return [];
}
}
function readLastRunDiagnostics(workspacePath: string): string[] {
const diagnosticsPath = join(workspacePath, 'logs', 'last-run-diagnostics.json');
if (!existsSync(diagnosticsPath)) return [];
try {
const data = JSON.parse(readFileSync(diagnosticsPath, 'utf-8')) as {
status?: string;
abortReason?: string | null;
finalOutput?: string;
movementHistory?: Array<{
name?: string;
next?: string | null;
toolsUsed?: string[];
outputPreview?: string;
}>;
contextActions?: unknown[];
};
const lines: string[] = [];
if (data.status || data.abortReason) {
lines.push(`- 前回ステータス: ${data.status ?? 'unknown'}${data.abortReason ? ` (${data.abortReason})` : ''}`);
}
for (const movement of data.movementHistory ?? []) {
const tools = movement.toolsUsed && movement.toolsUsed.length > 0
? ` tools=${movement.toolsUsed.join(',')}`
: '';
lines.push(`- movement ${movement.name ?? 'unknown'} -> ${movement.next ?? 'unknown'}${tools}`);
if (movement.outputPreview) {
lines.push(` - output: ${truncateRetryText(movement.outputPreview, 300)}`);
}
}
if (data.finalOutput) {
lines.push(`- 最終出力プレビュー: ${truncateRetryText(data.finalOutput, RETRY_DIAGNOSTICS_PREVIEW_LENGTH)}`);
}
if (data.contextActions && data.contextActions.length > 0) {
lines.push(`- context action: ${JSON.stringify(data.contextActions.slice(-3))}`);
}
return lines;
} catch {
return [];
}
}
/**
* Effective space id for per-space MCP/SSH tool resolution.
*
* A case-space task already carries a real `space_id` → use it. A
* personal-workspace task carries `space_id=NULL`, but the owner's MCP/SSH
* servers are registered against the owner's real personal-space id; resolve
* NULL+owner → that id (mirroring how `resolveTaskFolderContext` maps NULL to
* the personal space for FOLDERS). No-auth jobs (no owner) stay NULL-space.
* The resolved id flows into `ToolContext.spaceId`, which BOTH the tool
* enumeration (`getToolDefs`) and the execute-time guard (`aggregator.executeTool`)
* read — so they always agree.
*/
export async function resolveToolSpaceId(
rawSpaceId: string | undefined,
ownerId: string | null | undefined,
resolvePersonalSpaceId: (ownerId: string) => Promise<string>,
): Promise<string | undefined> {
if (rawSpaceId !== undefined) return rawSpaceId;
if (!ownerId) return undefined;
try {
return await resolvePersonalSpaceId(ownerId);
} catch {
// Fail CLOSED: a task that SHOULD be scoped to its owner's personal space
// must never fall back to the global NULL-space MCP/SSH set on a transient
// resolver error (that would leak NULL-space servers across the boundary).
// An owner-scoped sentinel matches no real space → zero per-space tools.
return `personal-unresolved:${ownerId}`;
}
}
export function buildRetryHandoffSummary(params: {
workspacePath: string;
job: Job;
errorMsg: string;
nextRetryAt?: string | null;
disposition: 'requeued_unhealthy' | 'retry' | 'failed';
}): string {
const lines: string[] = [
'# Retry Handoff',
'',
`Generated: ${new Date().toISOString()}`,
`Job: ${params.job.id}`,
`Disposition: ${params.disposition}`,
`Attempt: ${params.job.attempt}/${params.job.maxAttempts}`,
];
if (params.nextRetryAt) lines.push(`Next retry at: ${params.nextRetryAt}`);
lines.push('', '## 失敗理由', truncateRetryText(params.errorMsg, 2_000));
const diagnostics = readLastRunDiagnostics(params.workspacePath);
if (diagnostics.length > 0) {
lines.push('', '## 前回実行の要約', ...diagnostics);
}
const lessons = readRetryLessons(params.workspacePath);
if (lessons.length > 0) {
lines.push('', '## これまでの lessons', ...lessons);
}
lines.push(
'',
'## 次のエージェントへの指示',
'- 前回の失敗理由と movement の進捗を踏まえ、同じ探索や同じ失敗を繰り返さないこと。',
'- 既に完了している作業・生成済みファイル・確認済み事項は再実行前に workspace とログで確認すること。',
'- 必要な情報が不足している場合は、全体を読み直すのではなく targeted Read / Grep / Bash で範囲を絞ること。',
);
return `${truncateRetryText(lines.join('\n'), RETRY_HANDOFF_MAX_LENGTH)}\n`;
}
function writeRetryHandoffSummary(params: {
workspacePath: string | null | undefined;
job: Job;
errorMsg: string;
nextRetryAt?: string | null;
disposition: 'requeued_unhealthy' | 'retry' | 'failed';
}): void {
if (!params.workspacePath) return;
try {
const logsDir = join(params.workspacePath, 'logs');
mkdirSync(logsDir, { recursive: true });
const summary = buildRetryHandoffSummary({
workspacePath: params.workspacePath,
job: params.job,
errorMsg: params.errorMsg,
nextRetryAt: params.nextRetryAt,
disposition: params.disposition,
});
writeFileSync(join(logsDir, 'retry-summary.md'), summary, 'utf-8');
} catch (err) {
logger.warn(`[worker] failed to write retry handoff summary: ${err}`);
}
}
function buildRetryHandoffContext(workspacePath: string, job: Job): string {
if (job.attempt <= 1 && !job.errorSummary) return '';
const summaryPath = join(workspacePath, 'logs', 'retry-summary.md');
if (!existsSync(summaryPath)) return '';
try {
const summary = truncateRetryText(readFileSync(summaryPath, 'utf-8'), RETRY_HANDOFF_MAX_LENGTH);
if (!summary) return '';
return [
'## Retry 復帰用引き継ぎ',
'このジョブは前回実行からの retry / 再キューです。以下を前提に、重複作業を避けて継続してください。',
'',
summary,
].join('\n');
} catch {
return '';
}
}
/** DB-level terminal statuses for a subtask job (distinct from piece-result statuses). */
const SUBTASK_DB_TERMINAL: readonly string[] = ['succeeded', 'failed', 'cancelled'];
/**
* Cancelable sleep: resolves after `ms`, but bails early (polling every ~100ms)
* when cancelCheck() returns true. Used for the SpawnSubTask stagger so a
* cancelled parent does not sit sleeping. Not a completion-wait.
*/
async function sleepWithCancel(ms: number, cancelCheck?: () => boolean): Promise<void> {
const step = 100;
let waited = 0;
while (waited < ms) {
if (cancelCheck?.()) return;
const chunk = Math.min(step, ms - waited);
await new Promise<void>((resolve) => setTimeout(resolve, chunk));
waited += chunk;
}
}
export async function maybeEnqueueReflection(
repo: Repository,
job: Job,
outcome: 'succeeded' | 'failed' | 'aborted',
cfg: Pick<ReflectionConfig, 'enabled' | 'workerRequired' | 'perUserDailyBudgetTokens'>,
workers: WorkerDef[] = [],
): Promise<void> {
if (!cfg.enabled) return;
if (job.taskKind === 'reflection') return;
// No-auth mode runs every job with ownerId=null. Reflection is per-user
// (memory/pieces live under data/users/{userId}/), so fall back to the same
// 'local' namespace the rest of the no-auth path uses (ToolContext, pieces,
// user-folder). Without this the enqueue gate skipped forever and reflection
// silently never ran in no-auth deployments.
const reflectionOwner = job.ownerId ?? 'local';
// worker_required enforcement: when true, at least one worker must have 'reflection' in its roles
if (cfg.workerRequired) {
const hasReflectionWorker = workers.some(
(w) => Array.isArray(w.roles) && w.roles.includes('reflection'),
);
if (!hasReflectionWorker) {
logger.warn(`[reflection] enqueue skipped reason=no_reflection_worker user=${reflectionOwner}`);
return;
}
}
// Per-user daily token budget check.
// Cap=0 means "no limit" — useful for fresh installs that haven't tuned the budget yet.
const cap = cfg.perUserDailyBudgetTokens ?? 0;
if (cap > 0) {
// Compute today's start in UTC (00:00:00.000 UTC).
const now = new Date();
const todayStartMs = Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), now.getUTCDate());
const metrics = repo.aggregateReflectionMetrics(reflectionOwner, todayStartMs);
const spent = metrics.tokensIn + metrics.tokensOut;
if (spent >= cap) {
const spentM = (spent / 1_000_000).toFixed(1);
const capM = (cap / 1_000_000).toFixed(1);
logger.info(`[reflection] enqueue skipped reason=budget user=${reflectionOwner} spent=${spentM}M cap=${capM}M`);
return;
}
}
const payload = JSON.stringify({
originalJobId: job.id,
userId: reflectionOwner,
pieceName: job.pieceName,
outcome,
// Space-as-Principal: a CASE-space job's learnings (memory/pieces) belong to
// that space, not the triggering user's personal folder. Carry the original
// job's space so the reflection runner resolves storage to data/spaces/{id}.
// NULL (personal / no-space) → unchanged (data/users/{userId}).
spaceId: job.spaceId ?? null,
});
await repo.createJob({
repo: `local/reflection-${job.id}`,
issueNumber: 0,
instruction: '',
pieceName: 'reflection',
role: 'reflection',
ownerId: reflectionOwner,
visibility: 'private',
taskKind: 'reflection',
payload,
} as any);
logger.info(`[reflection] enqueued original=${job.id} owner=${reflectionOwner} piece=${job.pieceName} outcome=${outcome}`);
}
export class Worker {
private running = false;
private inflight = 0;
private polling = false;
private stopped = false;
private pollInterval: ReturnType<typeof setInterval> | null = null;
private healthInterval: ReturnType<typeof setInterval> | null = null;
/** Live sibling list, injected by WorkerManager for idle-preferring claims. */
private siblingsAccessor: (() => Worker[]) | null = null;
/** Last initialize() result — whether we'd actually claim if poked. */
private lastAvailable = false;
/** Job id we deferred to an idler last round (safety net against stuck yields). */
private lastYieldedJobId: string | null = null;
private workerId: string;
private endpoint: string;
private model: string | undefined;
private availableModels: Set<string> = new Set();
private healthy = false;
private lastHealthError: string | null = null;
private contextLimitTokens: number = 128_000;
private mcpTokenManager: McpTokenManager | null = null;
/**
* Phase 3b: optional Prometheus metrics handle. When set, the worker
* emits jobs_total / active_jobs / job_duration_seconds in
* executeJob's start/finally, llm_calls_total via the AgentLoop
* onLLMCall callback, and tool_calls_total via the new onToolMetric
* callback in agent-loop.ts. Wired by WorkerManager after the metric
* registry exists.
*/
private workerMetrics: import('./metrics/worker-metrics.js').WorkerMetrics | null = null;
private skillCatalog: import('./engine/skills.js').SkillCatalog | null = null;
/**
* V2 Web Push notification service. Null when push is disabled via
* config or when the worker is built without one (tests). Hooks fire
* via enqueue (fire-and-forget) so a slow push service can't block job
* execution.
*
* Spec: docs/superpowers/specs/2026-05-28-browser-notifications-v2-webpush.md.
*/
private pushService: import('./push-service.js').PushService | null = null;
constructor(
workerId: string,
endpoint: string,
model: string | undefined,
private repo: Repository,
private config: AppConfig,
) {
this.workerId = workerId;
this.endpoint = endpoint;
this.model = model;
}
public setMcpTokenManager(tm: McpTokenManager | null): void {
this.mcpTokenManager = tm;
}
public setSkillCatalog(catalog: import('./engine/skills.js').SkillCatalog): void {
this.skillCatalog = catalog;
}
public setPushService(svc: import('./push-service.js').PushService | null): void {
this.pushService = svc;
}
/**
* Hot-swap the global config on a still-running worker. Used by
* WorkerManager's differential rebuild: when a config change does NOT
* touch this worker's own def (e.g. a tool size limit changed), we keep
* the worker — and any in-flight job — alive and just refresh the config
* it reads for future jobs. Def-derived values (roles, endpoint,
* maxConcurrency) are read live via getWorkerDef(), so they stay correct.
* In-flight jobs keep the settings they captured at start, by design.
*/
public updateConfig(config: AppConfig): void {
this.config = config;
}
/** Jobs currently executing in this worker's detached loops. */
public get inflightCount(): number {
return this.inflight;
}
/** Free execution slots right now (max_concurrency inflight). */
public get freeSlots(): number {
return Math.max(0, this.getMaxConcurrency() - this.inflight);
}
/** True when this worker would actually pick up a job if poked. */
public get availableForClaim(): boolean {
return this.running && !this.stopped && this.lastAvailable
&& isExecutionWorker(this.getWorkerDef());
}
/** Whether this worker serves jobs of the given role. */
public canClaimRole(role: string): boolean {
return this.supportsRole(role);
}
/** Nudge this worker to poll immediately (hands a yielded job to an idler). */
public pokePoll(): void {
void this.processNext();
}
/** WorkerManager injects the live sibling list so claims prefer idler workers. */
public setSiblingsAccessor(fn: () => Worker[]): void {
this.siblingsAccessor = fn;
}
/**
* Find the idlest sibling that has strictly more free slots than us and
* serves `role`. Returns null when we are (tied for) the most free, in which
* case we should claim the job ourselves.
*/
private findIdlerCompetitor(role: string): Worker | null {
const others = (this.siblingsAccessor?.() ?? []).filter((s) => s !== this);
const idx = pickIdlerIndex(
this.freeSlots,
others.map((s) => ({
freeSlots: s.freeSlots,
availableForClaim: s.availableForClaim,
servesRole: s.canClaimRole(role),
})),
);
return idx >= 0 ? others[idx]! : null;
}
/**
* Fire a V2 push for a job status transition. Fire-and-forget — never
* throws and never awaits the underlying queue. Skips silently when
* - push is disabled (pushService === null)
* - the job has no owner (legacy / system-issued)
* - the job is not a local task (sub-task pushes go to the parent owner;
* we still send for direct local tasks).
* Reflection jobs are also skipped — they're an internal mechanism, not
* user-facing work.
*/
private enqueuePush(
job: Job,
event: 'running' | 'succeeded' | 'failed' | 'waiting_human',
): void {
if (!this.pushService) return;
if (job.taskKind === 'reflection') return;
if (!job.ownerId) return;
const localTaskId = getLocalTaskId(job.repo);
if (localTaskId === null) return;
// Title lookup is cheap (single-row SELECT) and synchronous via the
// raw db handle. Falling back to a generic label is fine — the
// push-service uses privacy-default payloads unless the user opted
// in via include_details.
let taskTitle = `Task #${localTaskId}`;
try {
const row = this.repo.getDb()
.prepare('SELECT title FROM local_tasks WHERE id = ?')
.get(localTaskId) as { title: string | null } | undefined;
if (row?.title) taskTitle = row.title;
} catch {
// best-effort; fall through with default title
}
this.pushService.enqueue({
event,
taskId: localTaskId,
taskTitle,
pieceName: job.pieceName,
ownerId: job.ownerId,
});
}
/**
* Phase 3b: install (or remove) the Prometheus metrics handle.
* Idempotent — calling with the same handle twice is fine. Null
* clears the handle, useful when reconfiguring at runtime.
*/
public setWorkerMetrics(
metrics: import('./metrics/worker-metrics.js').WorkerMetrics | null,
): void {
this.workerMetrics = metrics;
}
private getWorkerDef(): WorkerDef {
const workerDef = this.config.provider.workers.find((worker) => worker.id === this.workerId);
if (!workerDef) {
throw new Error(`Worker config not found: ${this.workerId}`);
}
return workerDef;
}
private getSupportedRoles(): string[] {
return this.getWorkerDef().roles ?? ['auto', 'fast', 'quality'];
}
private getMaxConcurrency(): number {
return Math.max(1, this.getWorkerDef().maxConcurrency ?? 1);
}
/**
* Hard wall-clock ceiling (ms) for a single LLM call, passed to every
* OpenAICompatClient. `undefined` lets the client default (2× idle timeout)
* apply; `provider.max_stream_minutes: 0` explicitly disables the cap.
*/
private resolveMaxStreamMs(): number | undefined {
return resolveMaxStreamMs(this.config.provider);
}
/** Hard wall-clock ceiling (ms) for one job's active execution; 0 disables. */
private resolveMaxJobMs(): number {
const mins = this.config.safety?.maxJobMinutes ?? DEFAULT_MAX_JOB_MINUTES;
return Math.max(0, mins) * 60 * 1000;
}
/**
* Start the per-job guard timer: polls for a cancel request (firing the
* abort mid-stream, which cancelCheck alone can't do because it only runs
* between agent-loop iterations) and enforces the hard execution deadline.
* Extracted so the timing logic is unit-testable. Returns the interval
* handle; the caller clears it in executeJob's finally.
*/
private startJobGuard(
jobId: string,
startedAt: number,
maxJobMs: number,
cancelCheck: () => boolean,
abortController: AbortController,
): ReturnType<typeof setInterval> {
return setInterval(() => {
try {
if (cancelCheck()) return; // 中断検知abort 済み
if (shouldDeadlineAbort(Date.now() - startedAt, maxJobMs, abortController.signal.aborted)) {
logger.warn(`[worker:${this.workerId}] job ${jobId} exceeded maxJobMinutes=${Math.round(maxJobMs / 60000)}; force-aborting to release slot`);
// Distinguishable reason so the loop classifies this as a deadline
// timeout (not a user cancel). See src/engine/cancellation.ts.
abortController.abort(DEADLINE_ABORT_REASON);
}
} catch { /* ignore */ }
}, JOB_GUARD_POLL_MS);
}
async initialize(): Promise<boolean> {
const workerDef = this.getWorkerDef();
const enabled = workerDef.enabled !== false;
if (!enabled) {
await this.repo.upsertWorkerNode({
workerId: this.workerId,
endpoint: this.endpoint,
enabled: false,
healthy: false,
roles: this.getSupportedRoles(),
availableModels: [],
inflightJobs: this.inflight,
maxConcurrency: this.getMaxConcurrency(),
lastError: 'disabled by config',
});
this.healthy = false;
this.lastHealthError = 'disabled by config';
logger.info(`[worker:${this.workerId}] disabled by config; skipping polling`);
return false;
}
try {
const ollamaBase = this.endpoint.replace(/\/v1\/?$/, '');
// Try Ollama /api/tags first, then fall back to OpenAI-compatible /v1/models.
//
// Forward `Authorization: Bearer <apiKey>` when the worker has one
// configured. The discovery probes (/api/tags / /v1/models) were
// previously sent un-authenticated, which caused 30s-interval 401
// floods against AAO Gateway endpoints (gateway requires Bearer auth
// on /v1/models) — the worker then logged "failed to fetch model
// list" indefinitely and `availableModels` stayed empty.
// Discovered during 2026-05-20 dogfooding on production aao.
const apiKey = this.getWorkerDef().apiKey;
const init: RequestInit = apiKey
? { headers: { Authorization: `Bearer ${apiKey}` } }
: {};
let models: string[] = [];
const ollamaRes = await fetch(`${ollamaBase}/api/tags`, init).catch(() => null);
if (ollamaRes?.ok) {
const data = await ollamaRes.json() as { models?: Array<{ name: string }> };
models = (data.models ?? []).map((m: { name: string }) => m.name);
} else {
const openaiBase = this.endpoint.replace(/\/?$/, '');
const openaiRes = await fetch(`${openaiBase}/models`, init).catch(() => null);
if (openaiRes?.ok) {
const data = await openaiRes.json() as { data?: Array<{ id: string }> };
models = (data.data ?? []).map((m: { id: string }) => m.id);
} else if (this.model) {
throw new Error(`failed to fetch model list from both /api/tags and /v1/models`);
}
// llama-server compat: model 未設定なら model 一覧 API は必須ではないので空配列で続行。
}
this.availableModels = new Set(models);
await this.repo.upsertWorkerNode({
workerId: this.workerId,
endpoint: this.endpoint,
enabled: true,
healthy: true,
roles: this.getSupportedRoles(),
availableModels: [...this.availableModels],
inflightJobs: this.inflight,
maxConcurrency: this.getMaxConcurrency(),
lastError: null,
});
if (!this.healthy || this.lastHealthError !== null) {
logger.info(`[worker:${this.workerId}] available models: ${[...this.availableModels].join(', ')}`);
}
this.healthy = true;
this.lastHealthError = null;
// Auto-detect context limit from Ollama if not configured
if (!this.config.context?.limitTokens) {
if (this.model) {
const contextLimit = await fetchOllamaContextLimit(this.endpoint, this.model);
if (contextLimit !== this.contextLimitTokens) {
logger.info(`[worker:${this.workerId}] context limit updated: ${contextLimit} tokens`);
this.contextLimitTokens = contextLimit;
}
} else {
// No model configured — try llama.cpp /props endpoint for context limit
const contextLimit = await fetchOllamaContextLimit(this.endpoint, '');
if (contextLimit !== this.contextLimitTokens) {
logger.info(`[worker:${this.workerId}] context limit updated: ${contextLimit} tokens`);
this.contextLimitTokens = contextLimit;
}
}
} else {
this.contextLimitTokens = this.config.context.limitTokens;
}
return true;
} catch (e) {
const errorMessage = e instanceof Error ? e.message : String(e);
this.availableModels.clear();
await this.repo.upsertWorkerNode({
workerId: this.workerId,
endpoint: this.endpoint,
enabled: true,
healthy: false,
roles: this.getSupportedRoles(),
availableModels: [],
inflightJobs: this.inflight,
maxConcurrency: this.getMaxConcurrency(),
lastError: errorMessage,
});
if (this.healthy || this.lastHealthError !== errorMessage) {
logger.warn(`[worker:${this.workerId}] failed to fetch model list: ${e}`);
}
this.healthy = false;
this.lastHealthError = errorMessage;
return false;
}
}
start(): void {
if (this.running) return;
this.running = true;
logger.info(`[worker:${this.workerId}] started`);
const tick = () => void this.processNext();
tick();
const baseInterval = 5000;
const jitter = Math.floor(Math.random() * 2000);
this.pollInterval = setInterval(tick, baseInterval + jitter);
const healthIntervalSeconds = Math.max(10, this.getWorkerDef().healthcheckIntervalSeconds ?? 30);
this.healthInterval = setInterval(() => void this.initialize(), healthIntervalSeconds * 1000);
}
stop(): void {
this.running = false;
this.stopped = true;
if (this.pollInterval) {
clearInterval(this.pollInterval);
this.pollInterval = null;
}
if (this.healthInterval) {
clearInterval(this.healthInterval);
this.healthInterval = null;
}
logger.info(`[worker:${this.workerId}] stopped`);
}
async waitForCompletion(timeoutMs = 30000): Promise<boolean> {
if (this.inflight === 0) return true;
const start = Date.now();
while (this.inflight > 0 && (Date.now() - start) < timeoutMs) {
await new Promise(resolve => setTimeout(resolve, 500));
}
return this.inflight === 0;
}
get id(): string { return this.workerId; }
private async processNext(): Promise<void> {
if (!isExecutionWorker(this.getWorkerDef()) || !this.running || this.stopped) return;
if (this.polling) return; // claim loop is single-flight (prevents over-claim)
this.polling = true;
try {
// スタックジョブ watchdog: LLM タイムアウトの2倍を閾値にする
try {
const staleMinutes = Math.max(20, (this.config.provider.timeoutMinutes ?? 10) * 2);
this.repo.recoverStuckRunningJobs(staleMinutes);
} catch (err) {
logger.warn(`[worker:${this.workerId}] recoverStuckRunningJobs error: ${err}`);
}
const available = await this.initialize();
this.lastAvailable = available;
if (!available) return;
const max = this.getMaxConcurrency();
while (this.inflight < max && this.running && !this.stopped) {
// Idle-preferring gate (most-free-wins): if a strictly-idler sibling
// serves the next job's role, hand it off (nudge that worker) instead
// of piling on. Safety net: if we already deferred this exact job last
// round and it is still here, the idler didn't take it (unhealthy /
// raced) — claim it ourselves so a job never gets stuck.
//
// Only consult the gate when there are sibling workers to defer to AND
// the repo supports peeking. Single-worker setups and unit tests skip
// it entirely — no extra query, no added latency, original claim timing.
const siblings = this.siblingsAccessor?.();
if (siblings && siblings.length > 1 && this.repo.peekNextClaimable) {
const peek = await this.repo.peekNextClaimable(this.workerId);
if (peek && peek.id !== this.lastYieldedJobId) {
const idler = this.findIdlerCompetitor(peek.requiredRole);
if (idler) {
this.lastYieldedJobId = peek.id;
idler.pokePoll();
break;
}
}
this.lastYieldedJobId = null;
}
// リトライジョブを優先
const job = await this.repo.claimNextRetryJob(this.workerId)
?? await this.repo.claimNextJob(this.workerId);
if (!job) break;
this.inflight++;
void this.runJobTracked(job); // 並行実行: await しない
}
} catch (err) {
logger.error(`[worker:${this.workerId}] processNext error: ${err}`);
} finally {
this.polling = false;
}
}
/** Run one job to completion, always restoring the inflight counter. */
private async runJobTracked(job: Job): Promise<void> {
try {
await this.executeJob(job);
} catch (err) {
logger.error(`[worker:${this.workerId}] runJobTracked error job=${job.id}: ${err}`);
} finally {
this.inflight--;
await this.reportInflight();
}
}
/** Push the live inflight count (and current health) to the worker_nodes row. */
private async reportInflight(): Promise<void> {
try {
await this.repo.updateWorkerNodeHealth(this.workerId, {
healthy: this.healthy,
lastError: this.lastHealthError,
inflightJobs: this.inflight,
availableModels: [...this.availableModels],
});
} catch (err) {
logger.warn(`[worker:${this.workerId}] reportInflight failed: ${err}`);
}
}
private supportsRole(role: string): boolean {
return this.getSupportedRoles().includes(role);
}
private buildLogMetadata(role: JobRole): ActivityLogMetadata {
return { workerId: this.workerId, mode: role };
}
/**
* サブタスクの ASK に対して、親ジョブの文脈を使って LLM に回答を生成させる
*/
private async answerSubtaskAsk(subtaskJob: Job, parentJobId: string, question: string): Promise<string> {
const parentJob = await this.repo.getJob(parentJobId);
const parentInstruction = parentJob?.instruction ?? '(不明)';
const workerDefForAnswer = this.getWorkerDef();
// Gateway routes by the subtask's tier; direct keeps the worker's model.
const resolvedModel = llmRoutingKey({
isGateway: workerDefForAnswer.proxy === true,
role: subtaskJob.requiredRole,
resolveDirectModel: () => this.model,
});
const timeoutMs = (this.config.provider.timeoutMinutes ?? 10) * 60 * 1000;
const llmClient = new OpenAICompatClient(
this.endpoint,
resolvedModel,
workerDefForAnswer.apiKey,
this.config.provider.retry,
timeoutMs,
this.contextLimitTokens,
this.config.safety?.promptGuardRatio,
undefined,
{ proxy: workerDefForAnswer.proxy === true, maxStreamMs: this.resolveMaxStreamMs() },
);
const messages: import('./llm/openai-compat.js').Message[] = [
{
role: 'system',
content: [
'あなたはタスクを管理する親エージェントです。',
'サブタスクがユーザーに確認を求めていますが、あなたが代わりに回答してください。',
'元の依頼の意図を汲み取り、サブタスクが作業を継続できるよう具体的に回答してください。',
'回答のみを簡潔に返してください。',
].join('\n'),
},
{
role: 'user',
content: [
'## 元の依頼',
parentInstruction,
'',
'## サブタスクの指示',
subtaskJob.instruction,
'',
'## サブタスクからの質問',
question,
].join('\n'),
},
];
let answer = '';
for await (const event of llmClient.chat(messages, undefined, undefined, { userId: parentJob?.ownerId ?? 'local' })) {
if (event.type === 'text') {
answer += event.text;
} else if (event.type === 'error') {
throw new Error(`LLM error: ${event.error}`);
}
}
return answer.trim() || '特に制約はありません。あなたの判断で進めてください。';
}
private writeRunDiagnostics(workspacePath: string, result: PieceRunResult): void {
try {
const logsDir = join(workspacePath, 'logs');
mkdirSync(logsDir, { recursive: true });
const diagnostics = {
generatedAt: new Date().toISOString(),
status: result.status,
abortReason: result.abortReason ?? null,
finalOutput: result.finalOutput,
movementHistory: result.movementHistory.map((entry) => ({
name: entry.name,
next: entry.result.next,
toolsUsed: entry.result.toolsUsed,
outputPreview: entry.result.output.slice(0, 600),
})),
contextActions: result.contextActions,
};
writeFileSync(join(logsDir, 'last-run-diagnostics.json'), `${JSON.stringify(diagnostics, null, 2)}\n`, 'utf-8');
} catch (err) {
logger.warn(`[worker:${this.workerId}] failed to write run diagnostics: ${err}`);
}
}
/**
* Resolve the workspace path for a job and ensure the directory tree
* (input/output/logs + a git repo) exists. Persists the resolved path
* back to the job + local-task records so downstream consumers can find
* the workspace.
*
* Throws on jobs that are neither local tasks nor sub-tasks (the orchestrator
* doesn't currently spin its own workspaces for raw repo/issue jobs).
*/
private async prepareJobWorkspace(
job: Job,
isLocalTask: boolean,
isSubTask: boolean,
localTaskId: number | null,
): Promise<string> {
const { repo: repoName, issueNumber, id: jobId } = job;
// 永続/一時的の解決は local-tasks-api が作成時に local_tasks.workspace_path へ
// 保存済みpersistent=スペース共有 space/{id}/files、ephemeral=ephemeral/{taskId})。
// worker はそれを尊重する。未設定の旧タスク・スケジューラ生成等のみ従来の
// local/{taskId} へフォールバック。(以前はここで常に local/{taskId} を強制し、
// 保存済みパスを上書きしていたため永続スペースのワークスペースが実行時に
// 効かず、生成ファイルがタスクごとの local/{taskId} に散っていた。)
const localTask = isLocalTask && localTaskId !== null
? await this.repo.getLocalTask(localTaskId)
: null;
const workspacePath = isLocalTask
? (localTask?.workspacePath ?? join(this.config.worktreeDir, 'local', String(localTaskId)))
: isSubTask
? (job.worktreePath ?? join(this.config.worktreeDir, 'subtasks', jobId))
: join(this.config.worktreeDir, repoName, String(issueNumber));
if (isLocalTask) {
mkdirSync(workspacePath, { recursive: true });
mkdirSync(join(workspacePath, 'input'), { recursive: true });
mkdirSync(join(workspacePath, 'output'), { recursive: true });
mkdirSync(join(workspacePath, 'logs'), { recursive: true });
// readonly/ : ユーザーが「エージェントに編集されたくないファイル」を置く閲覧専用領域。
// 個人/local ワークスペースでも Files タブに最初から見えるよう作成する。
mkdirSync(join(workspacePath, 'readonly'), { recursive: true });
await ensureWorkspaceGitRepo(workspacePath);
// workspace_path が未設定の旧タスクのみ backfill保存済みは上書きしない
if (localTaskId !== null && !localTask?.workspacePath) {
await this.repo.updateLocalTask(localTaskId, { workspacePath });
}
} else if (isSubTask) {
// SpawnSubTask 経由で worktreePath が設定されている前提
if (!job.worktreePath) {
throw new Error(`Sub-task job ${jobId} has no worktreePath set`);
}
mkdirSync(job.worktreePath, { recursive: true });
mkdirSync(join(job.worktreePath, 'output'), { recursive: true });
mkdirSync(join(job.worktreePath, 'logs'), { recursive: true });
await ensureWorkspaceGitRepo(job.worktreePath);
} else {
throw new Error(`Unsupported job type: repo="${repoName}" is neither a local task nor a sub-task`);
}
await this.repo.updateJob(jobId, { worktreePath: workspacePath });
return workspacePath;
}
/**
* Run the two pre-execution gates: role capability check, issue lock.
* On failure, requeue the job (so another worker can pick it up) and
* return false so executeJob returns early.
*/
private async acquireJobOrRequeue(job: Job): Promise<boolean> {
const { repo: repoName, issueNumber, id: jobId } = job;
if (!this.supportsRole(job.requiredRole)) {
await this.repo.updateJob(jobId, { status: 'queued', workerId: null });
await this.repo.addAuditLog(jobId, 'job_requeued_capability_mismatch', 'worker', {
workerId: this.workerId,
requiredRole: job.requiredRole,
});
logger.info(`[worker:${this.workerId}] requeued job ${jobId} due to role mismatch (role=${job.requiredRole})`);
return false;
}
const locked = await this.repo.lockIssue(repoName, issueNumber, jobId);
if (!locked) {
await this.repo.updateJob(jobId, { status: 'queued', workerId: null });
await this.repo.addAuditLog(jobId, 'job_requeued_issue_locked', 'worker', {
workerId: this.workerId,
});
logger.info(`[worker:${this.workerId}] job ${jobId}: issue ${repoName}#${issueNumber} already locked, skipping`);
return false;
}
return true;
}
/**
* task_kind='reflection' の実行パスexecuteJob から委譲。workspace 準備と
* agent / LLM ループをバイパスして固定ハンドラ (handleReflectionJob) を走らせる。
* finally の health 更新 + issue unlock は agent パスの finally と同じ後始末。
*/
private async runReflectionJobPath(job: Job): Promise<void> {
const { repo: repoName, issueNumber } = job;
try {
await this.handleReflectionJob(job);
} finally {
await this.repo.updateWorkerNodeHealth(this.workerId, {
healthy: this.healthy,
lastError: this.lastHealthError,
inflightJobs: this.inflight,
availableModels: [...this.availableModels],
});
await this.repo.unlockIssue(repoName, issueNumber);
}
}
/**
* 進捗レポーターの runtime_dir 解決executeJob から切り出し)。
* 計画5 (Opus P1): reporter は logPath を構築時に固定するため、runtime_dir を
* 構築前に解決しておく必要がある。persistent スペースタスクは local_task の
* runtime_dirspace/{id}/runs/{taskId}を使い、readerfile API の logs
* セクション = logRoot(task))と同じ dir を指すことで writer/reader の
* split-brain を防ぐ。localTaskId / runtime_dir が無い ephemeral / legacy /
* gitea-issue では undefined となり、reporter は workspacePath/logs に
* フォールバックして従来と byte 一致する。
*/
private async resolveReporterRuntime(
job: Job,
localTaskId: number | null,
): Promise<{
localTaskForRuntime: { runtimeDir?: string | null; workspaceMode?: string | null; spaceId?: string | null } | undefined;
reporterRuntimeDir: ReturnType<typeof resolveReporterRuntimeDir>;
}> {
const jobId = job.id;
let localTaskForRuntime: { runtimeDir?: string | null; workspaceMode?: string | null; spaceId?: string | null } | undefined;
if (localTaskId != null) {
try {
const lt = await this.repo.getLocalTask(localTaskId);
localTaskForRuntime = lt ?? undefined;
} catch (err) {
logger.warn(`[worker:${this.workerId}] job ${jobId} getLocalTask for reporter runtime_dir failed, falling back to workspace/logs: ${err}`);
}
}
const reporterRuntimeDir = resolveReporterRuntimeDir({
jobRuntimeDir: job.runtimeDir,
localTaskRuntimeDir: localTaskForRuntime?.runtimeDir,
});
return { localTaskForRuntime, reporterRuntimeDir };
}
/**
* Agent 実行パスの space / folder / runtime コンテキスト解決executeJob から
* 切り出し。piece のロード元 custom ディレクトリ、conflict detection の要否、
* 実行ログ root、per-space MCP/SSH の実効 spaceId を確定する。ロジック・
* 条件・順序は移動のみで不変(元コメントも保持)。
*/
private async resolveRunSpaceContext(
job: Job,
localTaskForRuntime: { runtimeDir?: string | null; workspaceMode?: string | null; spaceId?: string | null } | undefined,
): Promise<{
folderContext: { rootDir: string; leafId: string } | undefined;
conflictDetection: boolean | undefined;
runtimeDir: string | undefined;
conflictDir: string | undefined;
spaceId: string | undefined;
customPieceDirs: string[];
}> {
const jobId = job.id;
// Piece 読み込み: per-user カスタムディレクトリ → global カスタムディレクトリ → builtin の順に探索
// No-auth jobs (ownerId null) resolve pieces from data/users/local/pieces, matching
// where no-auth POST now writes (LOCAL_OWNER='local' in pieces-api.ts).
const userFolderRoot = this.config.userFolderRoot ?? './data/users';
const ownerForPieces = job.ownerId ?? 'local';
// Spaces foundation: for LOCAL tasks, resolve the effective space folder
// (data/users/{owner} for personal / null space_id, or data/spaces/{id}
// for case spaces). Gitea-issue jobs (no local task id) stay undefined =>
// legacy personal-folder resolution (backward compatible).
let folderContext: { rootDir: string; leafId: string } | undefined;
const localTaskIdForSpace = getLocalTaskId(job.repo);
// Spaces foundation (plan 3): enable conflict detection only for persistent
// (shared) local-task workspaces. Ephemeral / gitea-issue jobs leave it OFF
// (undefined) → legacy byte-identical Read/Write/Edit behaviour.
let conflictDetection: boolean | undefined;
// 計画5: タスクの実行ログ root。persistent スペースタスクは local_task の
// runtime_dirspace/{id}/runs/{taskId}、subtask は親から継承した
// jobs.runtime_dir を使う。どちらも無ければ undefined runPiece / reporter が
// workspacePath/logs にフォールバックephemeral / legacy
let runtimeDir: string | undefined = job.runtimeDir ?? undefined;
// 計画5 (spec §10.3): 競合台帳 dir。スペースタスクは space/{id}/.conflict
// files の外)、それ以外は ToolContext 側で workspacePath/.conflict に
// フォールバック後方互換。undefined のまま渡してフォールバックさせる。
let conflictDir: string | undefined;
// Spaces foundation: the space this run belongs to. Threaded into
// ToolContext.spaceId and CONSUMED for per-space MCP/SSH resolution
// (Phases 2-3). Resolved from the local task's space_id (preferred), but
// INITIALIZED from the trusted jobs.space_id (set server-side at enqueue)
// so a getLocalTask exception below degrades to the job's own space —
// NEVER fails open to the global NULL-space MCP/SSH set (Opus security P2).
// Undefined only for gitea-issue jobs / genuinely space-less tasks.
let spaceId: string | undefined = job.spaceId ?? undefined;
if (localTaskIdForSpace != null) {
try {
folderContext = await this.repo.resolveTaskFolderContext(localTaskIdForSpace, userFolderRoot);
} catch (err) {
logger.warn(`[worker:${this.workerId}] job ${jobId} resolveTaskFolderContext failed, using legacy personal folder: ${err}`);
}
// 計画5 (Opus P1): reporter 構築時に取得済みの local_task 行を再利用する
// localTaskIdForSpace === localTaskId で同一行)。フェッチ失敗で未取得の
// 場合のみ防御的に再取得する。
try {
const localTaskForMode = localTaskForRuntime ?? await this.repo.getLocalTask(localTaskIdForSpace);
conflictDetection = localTaskForMode?.workspaceMode === 'persistent';
runtimeDir = localTaskForMode?.runtimeDir ?? undefined;
// Prefer the local task's space; fall back to the job's space_id.
spaceId = localTaskForMode?.spaceId ?? job.spaceId ?? undefined;
if (conflictDetection && localTaskForMode?.spaceId) {
conflictDir = spaceConflictDir(this.config.worktreeDir, localTaskForMode.spaceId);
}
} catch (err) {
logger.warn(`[worker:${this.workerId}] job ${jobId} getLocalTask for workspaceMode failed, conflict detection OFF: ${err}`);
}
}
// Personal-workspace tasks carry space_id=NULL, but the user registered
// their MCP/SSH servers against their real personal-space id. Mirror what
// resolveTaskFolderContext does for folders: resolve NULL+owner to the
// owner's personal space so per-space MCP/SSH enumeration AND the
// execute-time guard (aggregator.executeTool) both see those servers.
// Applied to EVERY job (not just the local-task branch) so spawned
// subtasks — which inherit the parent's NULL space and don't re-enter the
// block above — also resolve to the owner's personal space. Case-space
// jobs keep their id; no-auth jobs (no owner) stay NULL-space (legacy);
// a resolver failure fails closed (sentinel), never the global NULL set.
// See docs/superpowers/specs/2026-06-19-personal-space-mcp-resolution-design.md
spaceId = await resolveToolSpaceId(spaceId, job.ownerId, (o) =>
this.repo.ensurePersonalSpace(o).then((s) => s.id),
);
// Custom-piece load dir: prefer the resolved space folder when present.
const pieceRootDir = folderContext?.rootDir ?? userFolderRoot;
const pieceLeafId = folderContext?.leafId ?? ownerForPieces;
const customPieceDirs = [
userPiecesDir(pieceRootDir, pieceLeafId),
this.config.customPiecesDir,
].filter((d): d is string => !!d);
return { folderContext, conflictDetection, runtimeDir, conflictDir, spaceId, customPieceDirs };
}
/**
* Model-mismatch requeue gate (direct mode only — gateway routes by
* role, see shouldRequeueForModelMismatch). executeJob から切り出し。
* true を返したら呼び出し側は即 returnジョブは queued に戻し済み)。
*/
private async requeueOnModelMismatch(job: Job, piece: PieceDef): Promise<boolean> {
const jobId = job.id;
if (
shouldRequeueForModelMismatch({
isGateway: this.getWorkerDef().proxy === true,
pieceModel: piece.model,
availableModels: this.availableModels,
workerModel: this.model,
})
) {
await this.repo.updateJob(jobId, {
status: 'queued',
workerId: null,
errorSummary: `Required model ${piece.model} is not available on ${this.workerId}`,
});
await this.repo.addAuditLog(jobId, 'job_requeued_model_mismatch', 'worker', {
workerId: this.workerId,
requiredModel: piece.model,
availableModels: [...this.availableModels],
});
logger.info(`[worker:${this.workerId}] requeued job ${jobId} due to model mismatch (${piece.model})`);
return true;
}
return false;
}
/**
* MCP 認証ゲート: piece.required_mcp に記載されたサーバーのトークンがなければ park。
* executeJob から切り出し。wait_reason='mcp_auth_required' の永続化は
* job-recoveryトークン取得時の自動再キューの前提条件なので変更禁止。
* true を返したら呼び出し側は即 returnジョブは waiting_human で停車済み)。
*/
private async parkOnMissingMcp(job: Job, piece: PieceDef, localTaskId: number | null): Promise<boolean> {
const jobId = job.id;
const missingMcp = (piece.required_mcp ?? []).filter(
(serverId) => !this.mcpTokenManager || !this.mcpTokenManager.hasToken(job.ownerId ?? '', serverId),
);
if (missingMcp.length > 0) {
await this.repo.updateJob(jobId, {
status: 'waiting_human',
waitReason: 'mcp_auth_required',
resumeMovement: piece.initial_movement ?? null,
});
if (localTaskId !== null) {
await this.repo.addLocalTaskComment(
localTaskId,
'system',
`この piece は MCP サーバー「${missingMcp.join(', ')}」との連携が必要です。Settings → MCP 接続から連携してください。`,
'event',
);
}
logger.info(`[worker:${this.workerId}] mcp gate parked job=${jobId} missing=${missingMcp.join(',')}`);
return true;
}
return false;
}
/**
* このジョブ用の LLM クライアント構築executeJob から切り出し)。
* Gateway routes by role; direct resolves the worker's model. The
* resolver thunk runs only in direct mode (no auto-select via gateway).
*/
private createJobLlmClient(
job: Job,
piece: PieceDef,
reporter: LocalProgressReporter,
): { llmClient: OpenAICompatClient; isProxyWorker: boolean } {
const workerDefForLlm = this.getWorkerDef();
const isProxyWorker = workerDefForLlm.proxy === true;
const resolvedModel = llmRoutingKey({
isGateway: isProxyWorker,
role: job.requiredRole,
resolveDirectModel: () => this.resolveModel(piece),
});
const timeoutMs = (this.config.provider.timeoutMinutes ?? 10) * 60 * 1000;
const llmClient = new OpenAICompatClient(
this.endpoint,
resolvedModel,
workerDefForLlm.apiKey,
this.config.provider.retry,
timeoutMs,
this.contextLimitTokens,
this.config.safety?.promptGuardRatio,
(line) => reporter.reportPromptPreflight(line),
{
proxy: isProxyWorker,
maxStreamMs: this.resolveMaxStreamMs(),
requestPromptProgress: workerDefForLlm.returnProgress === true,
},
);
return { llmClient, isProxyWorker };
}
/**
* runPiece に渡す piece 実行オプションASK 再開・割り込み注入・SpawnSubTask・
* 未消化サブタスク検知の構築。executeJob から切り出しcopy-and-delete
* spawnSubTask 閉包の zombie guard / TOCTOU 再チェック / inheritJobContext に
* よる親→子継承所有権・可視性・スペース・browser_sessionは行単位で不変。
*/
private buildPieceRunOptions(args: {
job: Job;
workspacePath: string;
isLocalTask: boolean;
localTaskId: number | null;
isSubTask: boolean;
reporter: LocalProgressReporter;
jobAbortController: AbortController;
cancelCheck: () => boolean;
}) {
const { job, workspacePath, isLocalTask, localTaskId, isSubTask, reporter, jobAbortController, cancelCheck } = args;
const jobId = job.id;
return {
resumeMovement: job.resumeMovement ?? undefined,
askCount: job.askCount,
maxAskPerJob: this.config.ask.maxPerJob,
checkInterjections: isLocalTask && localTaskId !== null && !isSubTask
? async (movementName: string) => {
const comments = await this.repo.getUninjectedComments(localTaskId);
if (comments.length === 0) return [];
const injected = comments.map(c => ({ id: c.id, body: c.body }));
this.repo.markCommentsInjected(injected.map(c => c.id));
reporter.reportInterjectionAck(injected, movementName);
return injected;
}
: undefined,
spawnSubTask: job.subtaskDepth < this.config.subtasks.maxDepth
? async (params: { title: string; instruction: string; piece?: string }) => {
// Zombie guard: after a deadline/user abort — especially a ②B
// hard-kill that already ran cancelPendingChildren and marked the
// job terminal — a detached runPiece can still reach SpawnSubTask.
// Refuse to enqueue a NEW orphan child: nothing live will consume
// it, and it would burn a worker slot under an already-cancelled
// parent. The abort signal catches the in-grace window (terminal
// status not written yet); the status check catches post-hardkill.
if (!canSpawnSubtask(jobAbortController.signal.aborted, this.repo.getJobStatusSync(jobId) ?? '')) {
throw new Error('ジョブが終了またはキャンセルされたため、新しいサブタスクは作成できません。');
}
const subJobs = await this.repo.getSubJobs(jobId);
const subtaskIndex = subJobs.length + 1;
if (subJobs.length >= this.config.subtasks.maxPerParent) {
throw new Error(`サブタスク上限 (${this.config.subtasks.maxPerParent}) に達しました。これ以上のサブタスクは作成できません。`);
}
const subtaskWorkspace = join(workspacePath, 'subtasks', String(subtaskIndex));
mkdirSync(subtaskWorkspace, { recursive: true });
mkdirSync(join(subtaskWorkspace, 'output'), { recursive: true });
mkdirSync(join(subtaskWorkspace, 'logs'), { recursive: true });
// 計画5 (Opus P1): サブタスクは親と分離した自前の isolated worktree
// subtaskWorkspaceで実行されるため、その activity.log は共有されず
// resume poisoning の対象にならない。よって runtime_dir はネストせず
// 未設定のままにし、reporter を subtaskWorkspace/logs にフォールバック
// させる。これは subtask の readersubtask-activity-api /
// subtask-files-api が worktreePath/logs を読む)と byte 一致し、
// writer/reader の split-brain を防ぐ。
// 親ジョブの role を継承
const subJobInstruction = [
`ui_profile: ${job.requiredRole}`,
'',
`# ${params.title}`,
'',
params.instruction,
].join('\n');
const subJob = await this.repo.createJob({
repo: `subtask/${jobId}`,
issueNumber: subtaskIndex,
instruction: subJobInstruction,
pieceName: params.piece ?? 'general',
parentJobId: jobId,
subtaskDepth: job.subtaskDepth + 1,
maxAttempts: 2,
role: job.requiredRole,
// 親ジョブから所有権・可視性・スペース・ブラウザセッションのバインドを
// 継承するspec §5.7 のスペース継承 + 保存セッションの引き継ぎ)。
// browserSessionProfileId が無いとサブタスク内の BrowseWeb が
// 保存セッション (ログイン Cookie) を使えず未ログインでアクセスする。
...inheritJobContext(job),
// 計画5 (Opus P1): サブタスクは isolated worktree で実行され共有
// されないため runtime_dir はネストしないreporter は
// subtaskWorkspace/logs にフォールバックし reader と一致する)。
runtimeDir: null,
});
await this.repo.updateJob(subJob.id, { worktreePath: subtaskWorkspace });
// TOCTOU close: the abort/hard-kill can land between the entry
// guard and this insert — finalizeHardkill's cancelPendingChildren
// reads getSubJobs *before* this child row exists and so misses it.
// Re-check now: if the parent is no longer live, cancel the child we
// just created (it would otherwise run as an orphan under a
// cancelled parent) and refuse the spawn. finalizeHardkill writes
// the terminal status before cancelPendingChildren, so a status read
// here reliably reflects an in-flight hard-kill.
if (!canSpawnSubtask(jobAbortController.signal.aborted, this.repo.getJobStatusSync(jobId) ?? '')) {
await this.repo.updateJob(subJob.id, { status: 'cancelled' });
logger.info(`[worker:${this.workerId}] cancelled just-spawned sub-task #${subtaskIndex} job=${subJob.id} (parent no longer live)`);
throw new Error('ジョブが終了またはキャンセルされたため、新しいサブタスクは作成できません。');
}
logger.info(`[worker:${this.workerId}] spawned sub-task #${subtaskIndex} depth=${job.subtaskDepth + 1} job=${subJob.id}`);
// ④ spawn stagger: brief delay so a burst of N SpawnSubTask enqueues
// doesn't all land in the same instant and jump GPU-slot priority over
// other queued work. Throttle only — NOT a completion-wait. Honors
// cancel so a cancelled parent doesn't sit sleeping.
if (this.config.subtasks.spawnStaggerMs > 0) {
await sleepWithCancel(this.config.subtasks.spawnStaggerMs, cancelCheck);
}
return { jobId: subJob.id, subtaskIndex, workspacePath: subtaskWorkspace };
}
: undefined,
// WaitSubTask / auto-park support: report whether this job has unwaited
// pending children, and cancel them when the parent ends without consuming
// them (orphan prevention). Both use getSubJobs (latest revision per index).
hasPendingSubtasks: async (): Promise<boolean> => {
const subs = await this.repo.getSubJobs(jobId);
return subs.some((s) => !SUBTASK_DB_TERMINAL.includes(s.status));
},
};
}
/**
* Browser session profile bindingexecuteJob から切り出し)。
* If this job is bound to a browser_session_profile, decrypt the
* captured Playwright storageState and pass it into runPiece so
* BrowseWeb can inject it into BrowserContext. Owner-mismatch and
* expired-profile checks fail-fast before the agent loop starts
* (throw はそのまま executeJob の catch → scheduleRetryOrFail に伝播)。
*/
private resolveBrowserSessionBinding(
job: Job,
localTaskId: number | null,
): {
browserSessionState: object | undefined;
browserSessionProfileId: number | undefined;
browserSessionProfile: { loggedInSelector: string | null; loginUrlPatterns: string[] } | undefined;
onAuthExpired: ((profileId: number, reason: string) => void) | undefined;
} {
let browserSessionState: object | undefined;
let browserSessionProfileId: number | undefined;
let browserSessionProfile:
| { loggedInSelector: string | null; loginUrlPatterns: string[] }
| undefined;
let onAuthExpired:
| ((profileId: number, reason: string) => void)
| undefined;
if (job.browserSessionProfileId) {
const sessRepo = new BrowserSessionRepo(this.repo.getDb());
const profile = sessRepo.getProfileByIdUnsafe(job.browserSessionProfileId);
if (!profile) {
sessRepo.audit({
actorUserId: job.ownerId ?? null,
ownerId: null,
profileId: job.browserSessionProfileId,
action: 'use',
result: 'error',
reason: 'profile not found',
jobId: job.id,
});
throw new Error(`Browser session profile ${job.browserSessionProfileId} not found`);
}
// Fail-closed owner check: a job with null/missing ownerId must not
// be allowed to decrypt any profile, even if the profile id would
// otherwise resolve. Helper audits + throws on rejection. Extracted
// to src/engine/browser-session-auth.ts so the contract is unit
// tested in isolation from the Worker class.
assertProfileOwner(profile, job, sessRepo);
if (profile.status !== 'active' || !profile.encryptedStateBlob) {
sessRepo.audit({
actorUserId: job.ownerId, ownerId: profile.ownerId, profileId: profile.id,
action: 'use', result: 'error', reason: `status=${profile.status}`, jobId: job.id,
});
throw new Error(`AUTH_SESSION_EXPIRED: profile ${profile.id} status=${profile.status}`);
}
const masterKeyPath = this.config.secrets?.masterKeyPath ?? './data/secrets/master.key';
// Space-as-Principal P2: space-owned profiles decrypt under the space DEK
// (with owner-DEK fallback for pre-migration blobs); personal under owner DEK.
let stateJson: string;
try {
stateJson = decryptProfileState({ masterKeyPath, sessRepo }, {
ownerId: profile.ownerId,
spaceId: profile.spaceId,
encryptedStateBlob: profile.encryptedStateBlob,
});
} catch (e) {
sessRepo.audit({
actorUserId: job.ownerId,
ownerId: profile.ownerId,
profileId: profile.id,
action: 'decrypt',
result: 'error',
reason: `state decrypt failed: ${(e as Error).message}`,
jobId: job.id,
});
throw e;
}
browserSessionState = JSON.parse(stateJson) as object;
browserSessionProfileId = profile.id;
browserSessionProfile = {
loggedInSelector: profile.loggedInSelector,
loginUrlPatterns: profile.loginUrlPatterns,
};
onAuthExpired = (pid, reason) => {
sessRepo.markProfileStatus(pid, 'expired', reason);
sessRepo.audit({
actorUserId: job.ownerId, ownerId: profile.ownerId, profileId: pid,
action: 'expire', result: 'success', reason, jobId: job.id,
});
// Best-effort task-level notification. Subtask jobs and
// gitea-issue jobs may not have a numeric local_task id.
if (localTaskId !== null) {
this.repo.addLocalTaskComment(
localTaskId,
'agent',
`⚠️ Browser session "${profile.label}" expired: ${reason}. Re-login from Settings → Browser Sessions.`,
'progress',
).catch(() => { /* ignore — comment posting is best-effort */ });
}
};
sessRepo.audit({
actorUserId: job.ownerId, ownerId: profile.ownerId, profileId: profile.id,
action: 'use', result: 'success', jobId: job.id,
});
sessRepo.touchUsed(profile.id);
}
return { browserSessionState, browserSessionProfileId, browserSessionProfile, onAuthExpired };
}
/**
* LLM に渡す指示コンテキストの組み立てexecuteJob から切り出し)。
*
* Piece handoff: when this job continues an earlier one in the same
* local_task, agent-loop injects a "this is a continuation of piece X"
* block into the system prompt. We resolve the prev piece name + the
* most recent agent result/ask comment as the LLM-visible carry-over.
*
* 会話コンテキスト P3: この継続ジョブが transcript を再生する場合
* handoffContext あり && 再生可能ターンあり = piece-runner の再生条件と完全一致)、
* 再生した実ターンと重複するコメント要約ブロックを抑制する。transcriptPath は
* runPiece に渡す runtimeDir から算出するため piece-runner の logsRoot と一致し、
* 抑制判定と再生判定の split-brain が構造的に起きない。
*/
private async buildInstructionContext(
job: Job,
workspacePath: string,
localTaskId: number | null,
runtimeDir: string | undefined,
): Promise<{
handoffContext: import('./engine/agent-loop.js').HandoffContext | undefined;
transcriptPath: string;
enrichedInstruction: string;
}> {
const jobId = job.id;
const isLocalTask = localTaskId !== null;
let handoffContext: import('./engine/agent-loop.js').HandoffContext | undefined;
if (job.continuedFromJobId && isLocalTask && localTaskId !== null) {
const prevJob = await this.repo.getJob(job.continuedFromJobId);
if (prevJob) {
const prevResultComment = await this.repo.getLatestResultComment(localTaskId);
handoffContext = {
prevPiece: prevJob.pieceName,
prevResult: prevResultComment?.body ?? null,
};
} else {
logger.warn(`[worker:${this.workerId}] continued_from_job_id=${job.continuedFromJobId} not found for job ${jobId}; skipping handoff context`);
}
}
const transcriptPath = join(runtimeDir ?? join(workspacePath, 'logs'), 'transcript.jsonl');
const willReplayTranscript =
handoffContext !== undefined && Conversation.hasReplayableTranscript(transcriptPath);
let enrichedInstruction = `${buildTimeContextBlock()}${job.instruction}`;
if (isLocalTask) {
try {
const comments = await this.repo.listLocalTaskComments(localTaskId);
const outputFiles = this.listDir(join(workspacePath, 'output'));
const inputFiles = this.listDir(join(workspacePath, 'input'));
const contextBody = buildLocalConversationContext({
comments,
jobInstruction: job.instruction,
inputFiles,
outputFiles,
omitRecentConversation: willReplayTranscript,
});
enrichedInstruction = `${buildTimeContextBlock()}${contextBody}`;
} catch (err) {
logger.warn(`[worker:${this.workerId}] failed to build local context: ${err}`);
}
}
const retryHandoffContext = buildRetryHandoffContext(workspacePath, job);
if (retryHandoffContext) {
enrichedInstruction = `${enrichedInstruction}\n\n${retryHandoffContext}`;
}
return { handoffContext, transcriptPath, enrichedInstruction };
}
/**
* Parse per-task options from job payload (e.g. { options: { mcpDisabled, skillsDisabled } }).
* executeJob から切り出し。パース失敗は warn のみ(従来どおりジョブは続行)。
*/
private resolveJobPayloadOptions(job: Job): { mcpDisabled: boolean; skillsDisabled: boolean } {
const jobId = job.id;
let jobPayloadOptions: Record<string, unknown> = {};
if (job.payload) {
try {
const parsed = JSON.parse(job.payload) as Record<string, unknown>;
if (parsed?.options && typeof parsed.options === 'object' && !Array.isArray(parsed.options)) {
jobPayloadOptions = parsed.options as Record<string, unknown>;
}
} catch {
logger.warn(`[worker:${this.workerId}] job ${jobId} failed to parse payload JSON`);
}
}
const mcpDisabled = jobPayloadOptions.mcpDisabled === true;
const skillsDisabled = jobPayloadOptions.skillsDisabled === true;
return { mcpDisabled, skillsDisabled };
}
/**
* Resolve the job owner's role to gate admin-only tool actions
* (e.g. system-scope InstallSkill). Defaults to 'user' when the owner
* is unknown (no-auth mode) or the lookup fails. executeJob から切り出し。
*/
private resolveJobUserRole(job: Job): 'admin' | 'user' {
const jobId = job.id;
let userRole: 'admin' | 'user' = 'user';
if (job.ownerId) {
try {
const ownerRow = this.repo.getUserById(job.ownerId);
if (ownerRow?.role === 'admin') userRole = 'admin';
} catch (err) {
logger.warn(`[worker:${this.workerId}] job ${jobId} owner role lookup failed: ${(err as Error).message}`);
}
}
return userRole;
}
/**
* Workspace tool policy + SSH 接続スコープの解決executeJob から切り出し)。
*
* Workspace tool policy: resolve once per job, applied to every movement.
* Uses the already-resolved spaceId (personal-space NULL→real-id handled
* by resolveToolSpaceId in resolveRunSpaceContext). Fail-closed: any error →
* safe defaults (sensitive tools off). Never throws — a policy error must
* not crash a job.
*
* SSH workspace scoping: when the workspace policy enables the 'ssh' category,
* resolve the connection IDs available to this job's space. This space-scoped
* list is the SOLE SSH connection gate (piece-level allowed_ssh_connections was
* removed in PR B), mirroring how MCP uses listEnabledForSpace. spaceId is
* already personal-space-resolved (NULL→real-id), so personal-WS jobs find
* their connections via listBySpace(personalSpaceId). NEVER '*' wildcard —
* always an explicit id list so cross-space isolation is guaranteed.
*/
private async resolveWorkspaceToolContext(
job: Job,
spaceId: string | undefined,
): Promise<{
workspaceTools: { allowedTools: string[]; editAllowed: boolean };
workspaceSshConnections: string[] | undefined;
}> {
const jobId = job.id;
let workspaceTools: { allowedTools: string[]; editAllowed: boolean };
let parsedPolicy: import('./engine/workspace-tool-policy.js').WorkspaceToolPolicy;
try {
const policyJson = spaceId != null ? this.repo.getSpaceToolPolicy(spaceId) : null;
parsedPolicy = parseToolPolicy(policyJson);
workspaceTools = await resolveWorkspaceTools(parsedPolicy);
} catch (err) {
logger.warn(`[worker:${this.workerId}] job ${jobId} workspace tool policy resolution failed, using safe defaults: ${(err as Error).message}`);
parsedPolicy = {};
workspaceTools = await resolveWorkspaceTools({});
}
let workspaceSshConnections: string[] | undefined;
if (parsedPolicy.enabledSensitive?.includes('ssh') && spaceId != null) {
try {
const sshSub = getSshSubsystem();
if (sshSub !== null) {
workspaceSshConnections = sshSub.connectionRepo.listBySpace(spaceId).map((c) => c.id);
logger.debug(`[worker:${this.workerId}] job ${jobId} ssh workspace connections resolved count=${workspaceSshConnections.length} spaceId=${spaceId}`);
} else {
// SSH subsystem not initialised (e.g. test environment without bridge setup).
// Leave undefined → piece-runner treats it as [] (no connections → SSH rejects).
logger.debug(`[worker:${this.workerId}] job ${jobId} ssh enabled in policy but subsystem not initialised — no connections available`);
}
} catch (err) {
logger.warn(`[worker:${this.workerId}] job ${jobId} ssh connection resolution failed: ${(err as Error).message}`);
// Leave undefined → piece-runner treats it as [] (fail-closed, no connections).
}
}
return { workspaceTools, workspaceSshConnections };
}
/**
* runPiece の起動executeJob から切り出し。executeJob 側で解決済みの
* 実行コンテキストを束ねて runPiece の options に配線する。await はせず
* Promise を返す — 呼び出し側が ②B hard-kill レース (raceWithGrace) に
* かけるため。options の各項目・ゲート条件・コメントは行単位で不変。
*/
private startPieceRun(args: {
job: Job;
piece: PieceDef;
enrichedInstruction: string;
llmClient: OpenAICompatClient;
workspacePath: string;
callbacks: PieceRunCallbacks;
toolsConfig: AppConfig['tools'];
pieceOptions: ReturnType<Worker['buildPieceRunOptions']>;
cancelCheck: () => boolean;
jobAbortController: AbortController;
customPieceDirs: string[];
folderContext: { rootDir: string; leafId: string } | undefined;
spaceId: string | undefined;
sessionIdentity: Awaited<ReturnType<typeof resolveSessionTaskId>>;
conflictDetection: boolean | undefined;
runtimeDir: string | undefined;
conflictDir: string | undefined;
contextManager: ContextManager;
workerDef: WorkerDef;
handoffContext: import('./engine/agent-loop.js').HandoffContext | undefined;
transcriptPath: string;
isLocalTask: boolean;
localTaskId: number | null;
isSubTask: boolean;
parentJobId: string | null;
browserSessionState: object | undefined;
browserSessionProfileId: number | undefined;
browserSessionProfile: { loggedInSelector: string | null; loginUrlPatterns: string[] } | undefined;
onAuthExpired: ((profileId: number, reason: string) => void) | undefined;
mcpDisabled: boolean;
skillsDisabled: boolean;
userRole: 'admin' | 'user';
workspaceTools: { allowedTools: string[]; editAllowed: boolean };
workspaceSshConnections: string[] | undefined;
}): Promise<PieceRunResult> {
const {
job,
piece,
enrichedInstruction,
llmClient,
workspacePath,
callbacks,
toolsConfig,
pieceOptions,
cancelCheck,
jobAbortController,
customPieceDirs,
folderContext,
spaceId,
sessionIdentity,
conflictDetection,
runtimeDir,
conflictDir,
contextManager,
workerDef,
handoffContext,
transcriptPath,
isLocalTask,
localTaskId,
isSubTask,
parentJobId,
browserSessionState,
browserSessionProfileId,
browserSessionProfile,
onAuthExpired,
mcpDisabled,
skillsDisabled,
userRole,
workspaceTools,
workspaceSshConnections,
} = args;
const { id: jobId, issueNumber } = job;
return runPiece(piece, enrichedInstruction, llmClient, workspacePath, callbacks, toolsConfig, {
...pieceOptions,
cancelCheck,
abortController: jobAbortController,
safetyConfig: this.config.safety,
searchFilter: this.config.searchFilter,
customPiecesDir: customPieceDirs,
// Spaces foundation: resolved space folder for AGENTS.md / memory /
// custom-piece resolution. Undefined for gitea-issue jobs => legacy.
folderContext,
// Spaces foundation: space id for future per-space MCP/SSH resolution.
// Phase 1 no-op (no consumer reads ctx.spaceId yet).
spaceId,
// Per-space Python package overlay (admin-installed wheels). Resolve the
// stable `current` symlink path so Bash can bind it read-only + expose it
// on PYTHONPATH. undefined when the feature is disabled; the Bash tool
// also no-ops if the path does not exist yet (nothing installed).
pythonPackagesDir: (() => {
const pk = resolvePythonPackagesConfig(this.config.pythonPackages);
if (!pk.enabled) return undefined;
return spaceCurrentDir(pk.dir, spaceKeyFor(spaceId, sessionIdentity.userId ?? job.ownerId ?? 'local'));
})(),
// Spaces foundation (plan 3): conflict detection for persistent local-task
// workspaces. Undefined (OFF) for ephemeral / gitea-issue jobs => legacy.
conflictDetection,
// 計画5: 実行ログ root + 競合台帳 dir。スペースタスクのみ非 undefined。
// ephemeral / legacy は undefined workspacePath/logs・workspacePath/.conflict後方互換
runtimeDir,
conflictDir,
contextManager,
vlmEnabled: workerDef.vlm === true,
jobId,
handoffContext,
// When this run IS a subtask, pass parent identity + child workspace
// path so the runner records the lineage in its run-start event.
// Subtask workspaces follow `<parent>/subtasks/<N>` where N is the
// subtask job's issueNumber.
parentJobId: isSubTask && parentJobId ? parentJobId : undefined,
childWorkspaceRelative: isSubTask ? `subtasks/${issueNumber}` : undefined,
// Mission Brief: only wire IO when this run is bound to a local
// task (the brief is per-LocalTask, not per-job). Subtask runs
// and gitea-issue runs leave it unset → MissionUpdate degrades
// to a no-op and the system prompt MISSION block is skipped.
missionBrief: isLocalTask && localTaskId !== null && !isSubTask
? this.repo.makeMissionBriefIO(localTaskId)
: undefined,
// Conversation recall: bind SearchTaskConversation / ReadTaskConversation
// to this local task's comments + this run's transcript for top-level
// local tasks; sub-tasks get a transcript-only variant (see the isSubTask
// branch below). gitea-issue runs leave it unset so the tools degrade
// gracefully.
//
// SECURITY (adversarial review): bind the transcript ONLY to a per-task
// `runtimeDir` (space/{spaceId}/runs/{taskId}). The old
// `runtimeDir ?? workspacePath/logs` fallback is UNSAFE for persistent
// space tasks — workspacePath is the space-SHARED files tree, so a
// NULL-runtime_dir task would expose another (possibly different-user)
// task's transcript in the same space. When runtimeDir is unset we pass
// no transcript path: comment search (always task_id-scoped) still works,
// transcript search simply returns nothing rather than risk a leak.
taskConversation: isLocalTask && localTaskId !== null && !isSubTask
? this.repo.makeTaskConversationIO(
localTaskId,
runtimeDir ? join(runtimeDir, 'transcript.jsonl') : undefined,
)
// サブタスクは backing local_task を持たないcomments 無し)が、自分の
// transcript は subtaskWorkspace/logs に書かれる。上で算出済みの
// transcriptPathpiece-runner の logsRoot と同一式)をそのまま再利用し、
// writer/reader の drift を防ぐ。上の SECURITY 注記が禁じる workspacePath/logs
// フォールバックは「永続スペースの共有 workspacePath」が対象で、サブタスクの
// workspacePath は subtasks/<index> の隔離ディレクトリなので安全(他タスクの
// transcript には届かない。workspaceTaskSearch は配線しない(他タスク横断は
// サブタスクに開かない — 設計 §2 非ゴール)。
: isSubTask
? this.repo.makeTranscriptOnlyConversationIO(transcriptPath)
: undefined,
// Workspace-wide task search: same gate as taskConversation, plus an
// owner id (needed to scope the search to the owner's space). In
// no-auth mode job.ownerId is null, so this is undefined and the tool
// degrades to a no-context message rather than searching unscoped.
workspaceTaskSearch: isLocalTask && localTaskId !== null && !isSubTask && job.ownerId
? this.repo.makeWorkspaceTaskSearchIO(localTaskId, job.ownerId)
: undefined,
// File provenance: bind reads to THIS run's workspace_path (the scoping
// boundary — a shared space workspace is queried only for its own rows,
// never another workspace's / user's lineage) and a recorder bound to
// this task/job/space. Gated on a resolvable local task id (need a
// numeric created_by_task_id). Subtask / gitea-issue runs leave both
// unset → the agent tools degrade to a no-op and Write/Edit/Bash skip
// recording.
fileProvenance:
isLocalTask && localTaskId !== null
? this.repo.makeFileProvenanceIO(workspacePath)
: undefined,
recordFileProvenance:
isLocalTask && localTaskId !== null
? (ev) =>
this.repo.recordProvenance({
workspacePath,
relPath: ev.relPath,
event: ev.event,
sourceKind: ev.sourceKind,
checksum: ev.checksum ?? null,
taskId: localTaskId,
jobId,
spaceId: spaceId ?? null,
piece: ev.pieceName,
movement: ev.movementName,
})
: undefined,
taskId: sessionIdentity.taskId,
userId: sessionIdentity.userId,
// Tool-request mechanism: per-task grant overlay, read fresh each run so
// a resume after inline approval picks up the newly-granted tool.
grantedTools:
isLocalTask && localTaskId !== null ? this.repo.getGrantedTools(String(localTaskId)) : undefined,
// Workspace tool policy: pre-resolved once per job above (fail-closed).
// Drives every movement's effective allowed tools and edit flag.
workspaceTools,
// SSH workspace scoping: connection IDs available to this job's space.
// Undefined when ssh not enabled in policy (movement declaration is fallback).
workspaceSshConnections,
// Pause for inline approval only when a user is reachable: a local task
// that is not itself a subtask (subtasks/headless runs auto-proceed).
toolApprovalInteractive: isLocalTask && !isSubTask,
// Tool-request mechanism: bind the recorder to this run's task/job/space.
// piece + movement are injected upstream by piece-runner.
recordToolRequest: (req) =>
this.repo.recordToolRequest({
...req,
// Use the local task id (same key as granted_tools + the decide API)
// so the task-detail query and grant overlay line up.
taskId: localTaskId !== null ? String(localTaskId) : null,
jobId,
spaceId: spaceId ?? null,
}),
// Package-request mechanism (RequestPackage): bind the recorder +
// installed/declined state ONLY when the feature is enabled AND this run
// has a resolvable space. Approval installs into spaces.python_packages
// (keyed by space id) and the resumed job's overlay key is
// spaceKeyFor(spaceId) = spaceId, so the two must agree — hence the hard
// space requirement. No space => the tool reports "unavailable".
...(() => {
const pk = resolvePythonPackagesConfig(this.config.pythonPackages);
if (!pk.enabled || spaceId == null) return {};
// Installed set is name→spec (version-aware): a pinned request is only
// "already installed" on an exact spec match; a bare-name request
// matches any installed version.
let installedPackages: Array<{ name: string; spec: string }> = [];
try {
const raw = this.repo.getSpacePythonPackages(spaceId);
if (raw) {
const parsed = JSON.parse(raw) as { packages?: Array<{ name?: unknown; spec?: unknown }> };
if (Array.isArray(parsed.packages)) {
installedPackages = parsed.packages
.map((p) => (typeof p?.name === 'string' && typeof p?.spec === 'string' ? { name: p.name, spec: p.spec } : null))
.filter((p): p is { name: string; spec: string } => p !== null);
}
}
} catch {
// best-effort: a malformed desired-state must not fail the job.
}
// Names already declined for THIS job (deny/auto_deny) → RequestPackage
// short-circuits a re-request instead of re-parking (deny loop guard).
const declinedPackages = this.repo.listDeclinedPackageNamesByJob(jobId);
return {
installedPackages,
declinedPackages,
recordPackageRequest: (req: {
spec: string;
normalizedName: string;
reason?: string | null;
status?: 'pending' | 'approved' | 'denied' | 'auto_denied';
pieceName?: string | null;
movementName?: string | null;
}) =>
this.repo.recordPackageRequest({
...req,
taskId: localTaskId !== null ? String(localTaskId) : null,
jobId,
spaceId,
}),
};
})(),
browserSessionState,
browserSessionProfileId,
browserSessionProfile,
onAuthExpired,
ownerId: job.ownerId,
mcpConfig: mergeMcpConfig(this.config.mcp),
userRole,
skillCatalog: this.skillCatalog ?? undefined,
mcpDisabled,
skillsDisabled,
});
}
/**
* piece 実行の完了処理executeJob から切り出し。②B hard-kill レース →
* 実行診断の書き出し → handlePieceResultジョブライフサイクル遷移・park
* 永続化・reflection 起票を含む)→ メトリクス status 確定、の順序は不変。
*
* metricFinalStatus は setter 経由で書き込む: hard-kill 時は finalizeHardkill
* の【前】に 'cancelled' をセットfinalizeHardkill が throw しても finally の
* メトリクスが 'cancelled' で記録される従来挙動を保持)、通常時は
* handlePieceResult の【後】にセットhandlePieceResult が throw したら
* 'error' のまま catch に渡る従来挙動を保持。hard-kill 時はここで処理を
* 打ち切って戻る(呼び出し側 try ブロックの末尾なので従来の early return と
* 同一の後続フロー = finally のみ実行)。
*/
private async settlePieceRun(args: {
piecePromise: Promise<PieceRunResult>;
jobAbortController: AbortController;
job: Job;
reporter: LocalProgressReporter;
workspacePath: string;
isLocalTask: boolean;
isSubTask: boolean;
parentJobId: string | null;
setMetricFinalStatus: (s: JobMetricStatus) => void;
}): Promise<void> {
const { piecePromise, jobAbortController, job, reporter, workspacePath, isLocalTask, isSubTask, parentJobId, setMetricFinalStatus } = args;
const jobId = job.id;
// ②B deadline hard-kill safety net. Race the piece run against a grace
// timer that arms when the job's abort signal fires (deadline OR cancel).
// ②A wires the abort into the realistic blocking tools so a run normally
// unwinds cooperatively; if it does not within the grace window, hard-kill
// so the inflight slot is freed and the job reaches a terminal state. The
// detached run keeps executing in-process (accepted zombie residual).
// deadlineGraceSeconds <= 0 disables the ②B hard-kill fallback: await the
// run directly and rely on ②A's cooperative abort (matches the documented
// "0 disables" semantics, same convention as maxJobMinutes=0).
const graceSec = this.config.safety?.deadlineGraceSeconds ?? DEFAULT_DEADLINE_GRACE_SECONDS;
const raced = graceSec > 0
? await raceWithGrace(piecePromise, jobAbortController.signal, graceSec * 1000)
: { ok: true as const, value: await piecePromise };
if (!raced.ok) {
logger.warn(`[worker:${this.workerId}] job ${jobId} did not unwind within grace=${graceSec}s after abort; hard-killing to release slot (deadline=${raced.deadline})`);
setMetricFinalStatus('cancelled');
await this.finalizeHardkill(job, reporter, isSubTask, parentJobId, raced.deadline);
return;
}
const result = raced.value;
logger.info(`[worker:${this.workerId}] job ${jobId} runPiece done: status=${result.status}`);
this.writeRunDiagnostics(workspacePath, result);
await this.handlePieceResult(result, job, reporter, workspacePath, isLocalTask, isSubTask, parentJobId);
setMetricFinalStatus(mapPieceStatusToMetric(result.status));
}
private async executeJob(job: Job): Promise<void> {
const { repo: repoName, issueNumber, id: jobId } = job;
const localTaskId = getLocalTaskId(repoName);
const isLocalTask = localTaskId !== null;
const parentJobId = getSubTaskParentJobId(repoName);
const isSubTask = parentJobId !== null;
const logMetadata = this.buildLogMetadata(job.requiredRole);
if (!(await this.acquireJobOrRequeue(job))) return;
// Phase 3b: job-lifecycle metrics. Inc active_jobs at start; capture
// a terminal status + duration in the finally block. `profile` maps
// to the assigned required role (the multi-profile / multi-piece
// operator-facing dimension).
const metricPiece = job.pieceName ?? 'unknown';
const metricProfile = job.requiredRole ?? 'unknown';
const jobStartedAtMs = Date.now();
let metricFinalStatus: 'succeeded' | 'failed' | 'aborted' | 'cancelled' | 'waiting_human' | 'error' = 'error';
if (this.workerMetrics) {
try {
this.workerMetrics.activeJobs.labels({ piece: metricPiece, profile: metricProfile }).inc();
} catch { /* metrics never affect business logic */ }
}
await this.repo.updateWorkerNodeHealth(this.workerId, {
healthy: this.healthy,
lastError: this.lastHealthError,
inflightJobs: this.inflight,
availableModels: [...this.availableModels],
});
// claimNextJob がすでに status = 'running' にセット済み
await this.repo.addAuditLog(jobId, 'job_started', 'worker', {});
// V2 push: notify on the first time a job transitions queued→running.
// Retry runs are intentionally silent — V1's 4s debounce relied on this
// and we keep the same UX (one running notification, not one per retry).
if (job.attempt === 1) {
this.enqueuePush(job, 'running');
}
// Reflection jobs bypass workspace preparation and the agent / LLM loop.
// task_kind='agent' (default) keeps the pre-existing piece-runner path.
if (job.taskKind === 'reflection') {
await this.runReflectionJobPath(job);
return;
}
const workspacePath = await this.prepareJobWorkspace(job, isLocalTask, isSubTask, localTaskId);
// 進捗レポーター
// ローカルタスク・サブタスクともに activity.log を書き出す
// サブタスクでは isSubTask=true を渡し、DB コメント書き込みをスキップする
const { localTaskForRuntime, reporterRuntimeDir } = await this.resolveReporterRuntime(job, localTaskId);
const reporter = new LocalProgressReporter(this.repo, localTaskId ?? issueNumber, workspacePath, logMetadata, isSubTask, reporterRuntimeDir);
// 会話コンテキスト (enrichedInstruction) は handoffContext / runtimeDir 解決後に
// 組み立てる下記。P3 の要約抑制判定が piece-runner の再生条件と同一になるよう、
// handoffContext と runtimeDir が確定してから構築する必要があるため。
// watchdog 誤検知防止: runPiece 実行中に updated_at を定期更新
let heartbeatTimer: ReturnType<typeof setInterval> | undefined;
// 中断の即応 + ジョブ全体のハードデッドラインfix: runaway stream cancel/reaper
let jobGuardTimer: ReturnType<typeof setInterval> | undefined;
try {
const { folderContext, conflictDetection, runtimeDir, conflictDir, spaceId, customPieceDirs } =
await this.resolveRunSpaceContext(job, localTaskForRuntime);
logger.info(`[worker:${this.workerId}] job ${jobId} loadPiece piece=${job.pieceName} customDirs=[${customPieceDirs.join(', ') || 'none'}] piecesDir=pieces`);
const piece = loadPiece(job.pieceName, 'pieces', customPieceDirs);
if (await this.requeueOnModelMismatch(job, piece)) return;
if (await this.parkOnMissingMcp(job, piece, localTaskId)) return;
const { llmClient, isProxyWorker } = this.createJobLlmClient(job, piece, reporter);
// キャンセル用 AbortController
const jobAbortController = new AbortController();
// キャンセルチェック: DB のジョブ状態が 'cancelled' になっていたら中断する
const cancelCheck = (): boolean => {
const isCancelled = this.repo.getJobStatusSync(jobId) === 'cancelled';
if (isCancelled) {
jobAbortController.abort();
}
return isCancelled;
};
// ASK 再開の場合、resume_movement を使用
const pieceOptions = this.buildPieceRunOptions({
job,
workspacePath,
isLocalTask,
localTaskId,
isSubTask,
reporter,
jobAbortController,
cancelCheck,
});
// サブジョブの delegate イベントを root 親ジョブへ中継するための job id を解決する。
// サブタスクでない通常ジョブは undefined中継なし
const relayJobId = isSubTask ? await resolveRootJobId(this.repo, job) : undefined;
const callbacks = this.buildPieceCallbacks(
jobId,
reporter,
isLocalTask,
localTaskId,
workspacePath,
// Seed the backend tracker with whatever was already persisted
// for this job (e.g. on retry / resume from ASK). Only matters for
// proxy workers; direct workers never produce a backend event.
isProxyWorker ? (job.lastBackendId ?? null) : null,
llmClient,
logMetadata,
runtimeDir,
relayJobId,
);
// 開始コメント
await reporter.reportMovementStart(`${piece.name} タスク開始`);
// ── ジョブガード ───────────────────────────────────────────────
// cancelCheck はエージェントループのイテレーション先頭でしか呼ばれない。
// LLM が停止トークンを出さずにトークンを吐き続ける暴走に入ると、処理は
// ストリーム読み取りの中で止まり、次のイテレーションに到達しないため
// cancelCheck が二度と呼ばれず、中断ボタンが効かなくなる。さらにその
// ジョブの inflight スロットも解放されず、後続ジョブが永久に queued の
// まま動かなくなる。
// このタイマーで (1) cancel を定期的に確認して生成途中でも abort を
// 発火させ、(2) 実行時間がハードデッドラインを超えたら強制 abort して
// スロットを必ず解放する。abort は cancelSignal 経由で in-flight の
// fetch を中断し、ストリームがほどけて executeJob が戻る。
jobGuardTimer = this.startJobGuard(jobId, Date.now(), this.resolveMaxJobMs(), cancelCheck, jobAbortController);
// ContextManager 初期化
const contextManager = new ContextManager(this.config.context ?? {});
contextManager.setContextLimit(this.contextLimitTokens);
// Piece 実行(ハートビート開始: 5分ごとに updated_at を更新)
heartbeatTimer = setInterval(() => {
try { this.repo.touchJobUpdatedAt(jobId); } catch { /* ignore */ }
}, 5 * 60 * 1000);
// VLM 対応: worker の vlm=true なら vision 設定を worker 自身の endpoint/model で上書き
const workerDef = this.getWorkerDef();
const toolsConfig = workerDef.vlm
? { ...this.config.tools, visionBaseUrl: this.endpoint, visionModel: this.model }
: this.config.tools;
logger.info(`[worker:${this.workerId}] job ${jobId} runPiece start`);
// Browser session keying: resolve the session-task identity for the
// ToolContext. Threaded into BrowseWeb / InteractiveBrowse so each
// task gets its own noVNC session and the Captcha Pool stays
// separate from per-task sessions. Subtasks walk up to find the
// root local task ID.
const sessionIdentity = await resolveSessionTaskId(this.repo, job);
// ── Browser session profile binding ─────────────────────────────
const { browserSessionState, browserSessionProfileId, browserSessionProfile, onAuthExpired } =
this.resolveBrowserSessionBinding(job, localTaskId);
const { handoffContext, transcriptPath, enrichedInstruction } =
await this.buildInstructionContext(job, workspacePath, localTaskId, runtimeDir);
const { mcpDisabled, skillsDisabled } = this.resolveJobPayloadOptions(job);
const userRole = this.resolveJobUserRole(job);
const { workspaceTools, workspaceSshConnections } = await this.resolveWorkspaceToolContext(job, spaceId);
const piecePromise = this.startPieceRun({
job,
piece,
enrichedInstruction,
llmClient,
workspacePath,
callbacks,
toolsConfig,
pieceOptions,
cancelCheck,
jobAbortController,
customPieceDirs,
folderContext,
spaceId,
sessionIdentity,
conflictDetection,
runtimeDir,
conflictDir,
contextManager,
workerDef,
handoffContext,
transcriptPath,
isLocalTask,
localTaskId,
isSubTask,
parentJobId,
browserSessionState,
browserSessionProfileId,
browserSessionProfile,
onAuthExpired,
mcpDisabled,
skillsDisabled,
userRole,
workspaceTools,
workspaceSshConnections,
});
await this.settlePieceRun({
piecePromise,
jobAbortController,
job,
reporter,
workspacePath,
isLocalTask,
isSubTask,
parentJobId,
setMetricFinalStatus: (s) => { metricFinalStatus = s; },
});
} catch (err) {
const errorMsg = err instanceof Error ? err.message : String(err);
const errorStack = err instanceof Error && err.stack ? err.stack : '(no stack)';
if (errorMsg.includes('Piece not found')) {
logger.error(`[worker:${this.workerId}] job ${jobId} loadPiece failed piece=${job.pieceName} customPiecesDir=${this.config.customPiecesDir ?? 'none'} error=${errorMsg}`);
}
logger.error(`[worker:${this.workerId}] job ${jobId} failed: ${errorMsg}`);
// Always log the stack so opaque errors (e.g. SqliteError: FOREIGN KEY
// constraint failed) can be traced to the offending insert/update.
logger.error(`[worker:${this.workerId}] job ${jobId} stack: ${errorStack}`);
const retryDisposition = await this.scheduleRetryOrFail(job, errorMsg, workspacePath, 'worker_exception');
if (retryDisposition !== 'requeued_unhealthy') {
await reporter.reportError(errorMsg);
}
await this.repo.addAuditLog(jobId, 'job_error', 'worker', { error: errorMsg });
} finally {
if (heartbeatTimer) clearInterval(heartbeatTimer);
if (jobGuardTimer) clearInterval(jobGuardTimer);
// Phase 3b: emit job lifecycle counters + duration histogram. The
// active gauge always decrements (matching the inc at start) so a
// process restart can't leak active_jobs > 0 forever.
if (this.workerMetrics) {
try {
this.workerMetrics.activeJobs.labels({ piece: metricPiece, profile: metricProfile }).dec();
this.workerMetrics.jobsTotal.labels({ piece: metricPiece, status: metricFinalStatus, profile: metricProfile }).inc();
this.workerMetrics.jobDurationSeconds
.labels({ piece: metricPiece, status: metricFinalStatus, profile: metricProfile })
.observe((Date.now() - jobStartedAtMs) / 1000);
} catch { /* metrics never affect business logic */ }
}
await this.repo.updateWorkerNodeHealth(this.workerId, {
healthy: this.healthy,
lastError: this.lastHealthError,
inflightJobs: this.inflight,
availableModels: [...this.availableModels],
});
await this.repo.unlockIssue(repoName, issueNumber);
}
}
private buildPieceCallbacks(
jobId: string,
reporter: LocalProgressReporter,
isLocalTask: boolean,
localTaskId: number | null,
workspacePath: string,
/**
* Initial value of jobs.last_backend_id from the DB. Seeds the backend
* tracker (and the sticky-routing hint) so a resumed/retried job goes
* back to the backend that already holds its KV cache.
* Falsy/null = no backend resolved yet.
*/
initialLastBackendId: string | null = null,
/** LLM client of this job — receives the sticky-routing hint per switch. */
llmClient?: { setPreferredBackendId(backendId: string | null): void },
/**
* The reporter's metadata object (shared by reference): mutating
* `backendId` here makes every subsequent activity.log line carry
* `[backend:...]` so the Progress tab can show the physical backend
* behind a proxy worker.
*/
logMetadata?: ActivityLogMetadata,
/** 計画5: 実行ログ root。checklist 進捗コメントの読込先。
* 未指定時は workspacePath/logs後方互換。Phase A では常に undefined。 */
runtimeDir?: string,
/** Root local-task job id to relay delegate events to (subtask only).
* When set and different from jobId, delegate callbacks emit a second
* event tagged with originJobId so the parent SSE stream receives live
* delegate progress from sub-jobs. */
relayJobId?: string,
): PieceRunCallbacks {
let movementStartTime = Date.now();
const toolUsageCounts = new Map<string, number>();
// Backend tracker (follow-current semantics, 2026-06): persists
// jobs.last_backend_id whenever the resolved backend CHANGES so the UI
// (pet, badges) follows where the job actually runs. Switches are rare
// because the gateway honors the x-aao-preferred-backend sticky hint
// (KV-cache reuse) — they only happen when the preferred backend goes
// offline or saturates. The tracker still guarantees that a failed DB
// persist leaves the in-memory value unchanged so the next event
// retries. See src/worker/sticky-backend.ts.
const workerIdLocal = this.workerId;
const backendTracker = createStickyBackendResolver({
initial: initialLastBackendId,
persist: (backendId) => this.repo.updateJob(jobId, { lastBackendId: backendId }),
logger: {
debug: (m) => logger.debug(m),
info: (m) => logger.info(m),
warn: (m) => logger.warn(m),
},
workerId: workerIdLocal,
jobId,
});
// Seed the sticky-routing hint + activity-log backend tag from the DB
// value (resume/retry goes straight back to the cache-warm backend).
if (initialLastBackendId) {
llmClient?.setPreferredBackendId(initialLastBackendId);
if (logMetadata) logMetadata.backendId = initialLastBackendId;
}
// Phase 3b: local copy of the sticky backend so the LLM-call metric
// has a stable backend_id label even before the persist returns.
// Direct workers (non-proxy) never fire onBackendResolved, so we
// fall back to the worker id (`gpu-rtx-a`) as the backend identity.
let metricBackendId = initialLastBackendId ?? workerIdLocal;
const metricModel = this.model ?? 'unknown';
const metricsRef = this.workerMetrics;
// Pending tool calls awaiting result, keyed by callId.
// On onToolResult, we pair via callId and persist a single tool_call comment.
const pendingToolCalls = new Map<string, { name: string; args: Record<string, unknown>; movement: string }>();
let currentMovementForTools = '';
const ARG_PREVIEW_CAP = 8 * 1024;
const RESULT_PREVIEW_CAP = 16 * 1024;
const truncate = (s: string, cap: number): string =>
s.length > cap ? s.slice(0, cap) + `\n…[truncated ${s.length - cap} bytes]` : s;
// thinking イベントは reasoning チャンクごとに来るため SSE を throttle。
const THINKING_EMIT_THROTTLE_MS = 2_000;
let lastThinkingEmitAt = 0;
return {
onMovementStart: (name) => {
movementStartTime = Date.now();
toolUsageCounts.clear();
currentMovementForTools = name;
this.repo.updateJob(jobId, { currentMovement: name, currentActivity: null });
reporter.reportMovementStart(name);
},
onToolUse: (toolName, input, callId) => {
toolUsageCounts.set(toolName, (toolUsageCounts.get(toolName) ?? 0) + 1);
const summary = summarizeToolInput(toolName, input);
this.repo.updateJob(jobId, { currentActivity: `${toolName}: ${summary}`.slice(0, 200) });
reporter.reportToolUse(toolName, input);
if (callId) {
pendingToolCalls.set(callId, { name: toolName, args: input, movement: currentMovementForTools });
}
if (jobEventBus.hasListeners(jobId)) {
jobEventBus.emitJob(jobId, {
type: 'tool_use',
toolName,
toolInput: summary,
callId: callId ?? '',
});
}
},
onToolCallDelta: (callId, name, chunk) => {
if (jobEventBus.hasListeners(jobId)) {
jobEventBus.emitJob(jobId, { type: 'tool_use_delta', callId, name, chunk });
}
},
onText: (text) => {
if (jobEventBus.hasListeners(jobId)) {
jobEventBus.emitJob(jobId, { type: 'text', text });
}
},
onPromptProgress: (progress) => {
if (jobEventBus.hasListeners(jobId)) {
jobEventBus.emitJob(jobId, {
type: 'prompt_progress',
processed: progress.processed,
total: progress.total,
timeMs: progress.timeMs,
cache: progress.cache,
});
}
},
// ── LLM ライブ状態(観測性 Phase 1/2────────────────────────────
// 方針: ライブ表示は SSE 専用、DB (currentActivity) には低頻度の
// 節目送信開始・リトライ・復旧ステージだけ書く。thinking は
// 高頻度なので throttle した SSE のみで DB には一切書かない。
onLlmRequestStart: ({ movementName, iteration }) => {
this.repo.updateJob(jobId, { currentActivity: `LLM: 応答待ち (${movementName})`.slice(0, 200) });
if (jobEventBus.hasListeners(jobId)) {
jobEventBus.emitJob(jobId, { type: 'llm_state', phase: 'waiting', movement: movementName, iteration });
}
},
onLlmRetry: (info) => {
const reasonShort = info.reason.slice(0, 80);
this.repo.updateJob(jobId, {
currentActivity: `LLM: リトライ待ち ${info.attempt}/${info.maxAttempts} (${reasonShort})`.slice(0, 200),
});
if (jobEventBus.hasListeners(jobId)) {
jobEventBus.emitJob(jobId, {
type: 'llm_state',
phase: 'retrying',
attempt: info.attempt,
maxAttempts: info.maxAttempts,
reason: reasonShort,
errorClass: info.errorClass,
});
}
},
onThinking: ({ chars }) => {
const now = Date.now();
if (now - lastThinkingEmitAt < THINKING_EMIT_THROTTLE_MS) return;
lastThinkingEmitAt = now;
if (jobEventBus.hasListeners(jobId)) {
jobEventBus.emitJob(jobId, { type: 'llm_state', phase: 'thinking', chars });
}
},
onContextRecovery: ({ stage, detail }) => {
this.repo.updateJob(jobId, { currentActivity: `LLM: コンテキスト復旧中 (${stage})`.slice(0, 200) });
if (jobEventBus.hasListeners(jobId)) {
jobEventBus.emitJob(jobId, { type: 'llm_state', phase: 'recovering', stage, reason: detail?.slice(0, 120) });
}
},
onDelegateLifecycle: (info) => {
if (jobEventBus.hasListeners(jobId)) {
jobEventBus.emitJob(jobId, {
type: 'delegate_lifecycle',
delegateRunId: info.delegateRunId,
parentRunId: info.parentRunId,
depth: info.depth,
description: info.description,
status: info.status,
});
}
if (relayJobId && relayJobId !== jobId && jobEventBus.hasListeners(relayJobId)) {
jobEventBus.emitJob(relayJobId, {
type: 'delegate_lifecycle',
delegateRunId: info.delegateRunId,
parentRunId: info.parentRunId,
depth: info.depth,
description: info.description,
status: info.status,
originJobId: jobId,
});
}
},
onDelegateText: (delegateRunId, text) => {
if (jobEventBus.hasListeners(jobId)) {
jobEventBus.emitJob(jobId, { type: 'delegate_text', delegateRunId, text });
}
if (relayJobId && relayJobId !== jobId && jobEventBus.hasListeners(relayJobId)) {
jobEventBus.emitJob(relayJobId, { type: 'delegate_text', delegateRunId, text, originJobId: jobId });
}
},
onDelegateTool: (delegateRunId, toolName, summary) => {
if (jobEventBus.hasListeners(jobId)) {
jobEventBus.emitJob(jobId, { type: 'delegate_tool', delegateRunId, toolName, toolInput: summary });
}
if (relayJobId && relayJobId !== jobId && jobEventBus.hasListeners(relayJobId)) {
jobEventBus.emitJob(relayJobId, { type: 'delegate_tool', delegateRunId, toolName, toolInput: summary, originJobId: jobId });
}
},
onTextPreview: (movementName, preview) => {
reporter.reportAssistantPreview(movementName, preview);
},
onContextAction: (action) => {
reporter.reportContextAction(action);
},
onContextUpdate: (payload) => {
this.repo.updateJobContext(jobId, payload).catch(err => {
logger.warn(`[worker:${this.workerId}] failed to persist context for job ${jobId}: ${err}`);
});
},
onLLMCall: (info) => {
reporter.reportLLMCall(info);
if (metricsRef) {
try {
metricsRef.llmCallsTotal
.labels({ worker_id: workerIdLocal, backend_id: metricBackendId, model: metricModel })
.inc();
metricsRef.llmCallDurationSeconds
.labels({ worker_id: workerIdLocal, backend_id: metricBackendId, model: metricModel })
.observe(info.durationMs / 1000);
} catch { /* metrics best-effort */ }
}
},
onBackendResolved: (info) => {
// Phase 3b: update the backend id used for LLM-call metrics.
if (info.backendId) {
metricBackendId = info.backendId;
// Sticky routing: ask the gateway to keep using this backend on
// the next request (KV-cache affinity).
llmClient?.setPreferredBackendId(info.backendId);
// Tag subsequent activity.log lines with the physical backend so
// the Progress tab shows more than the proxy worker's name.
if (logMetadata) logMetadata.backendId = info.backendId;
}
// Fire-and-forget: agent-loop's onBackendResolved signature is
// sync (void). The tracker handles persist errors internally;
// we just attach a final guard to log any unexpected throw.
// cacheKey is observed but not persisted at the job level —
// Phase B's NodeStatusWidget will track cache hits out-of-band.
backendTracker.onEvent(info).catch(err => {
logger.warn(`[worker:${this.workerId}] backend tracker threw for job ${jobId}: ${err}`);
});
},
onMovementComplete: (movementName, result) => {
const durationMs = Date.now() - movementStartTime;
const tools: Record<string, number> = {};
for (const [name, count] of toolUsageCounts) {
tools[name] = count;
}
reporter.reportMovementComplete(movementName, result.output, result.next);
if (isLocalTask) {
const isTerminal = result.next === 'COMPLETE' || result.next === 'ABORT' || result.next === 'ASK';
const summary = !isTerminal ? (result.output?.trim() || undefined) : undefined;
const progressBody = JSON.stringify({ movement: movementName, tools, durationMs, ...(summary ? { summary } : {}) });
this.repo.addLocalTaskComment(localTaskId!, 'agent', progressBody, 'progress')
.catch(err => logger.warn(`[worker:${this.workerId}] failed to insert progress comment: ${err}`));
if (isTerminal && jobEventBus.hasListeners(jobId)) {
jobEventBus.emitJob(jobId, { type: 'done' });
}
}
},
onToolResult: (toolName, info, callId) => {
const { isError } = info;
reporter.reportToolResult(toolName, info);
// Pair with pending tool_use via callId, then persist as comment + emit SSE.
const pending = callId ? pendingToolCalls.get(callId) : undefined;
if (callId) pendingToolCalls.delete(callId);
if (isLocalTask && callId && pending) {
let argsStr: string;
try { argsStr = truncate(JSON.stringify(pending.args), ARG_PREVIEW_CAP); }
catch { argsStr = '"<unserializable>"'; }
const toolCallBody = JSON.stringify({
type: 'tool_call',
callId,
movement: pending.movement,
name: toolName,
args: argsStr,
result: truncate(info.result, RESULT_PREVIEW_CAP),
isError,
durationMs: info.durationMs,
cacheHit: info.cacheHit,
});
this.repo.addLocalTaskComment(localTaskId!, 'agent', toolCallBody, 'progress')
.catch(err => logger.warn(`[worker:${this.workerId}] tool_call comment failed: ${err}`));
}
if (jobEventBus.hasListeners(jobId) && callId) {
jobEventBus.emitJob(jobId, {
type: 'tool_result',
toolName,
callId,
toolOutput: truncate(info.result, 2 * 1024),
toolIsError: isError,
});
}
// Phase 3b: count every tool invocation. success label is the
// string form so Grafana queries can group by it. Same
// best-effort guard as the LLM emission above.
//
// Phase 3b post-review: normalize the tool_name label so a
// piece firing arbitrary mcp__*/user-defined names doesn't
// explode label cardinality. MCP tools collapse to a single
// `mcp` bucket; unknown names land in `unknown`. The full
// tool_name is still visible in the activity log + reporter,
// so the metric drop only affects Prometheus dimensions.
if (metricsRef) {
try {
metricsRef.toolCallsTotal
.labels({ tool_name: normalizeToolNameForMetric(toolName), success: isError ? 'false' : 'true' })
.inc();
} catch { /* metrics best-effort */ }
}
if (isLocalTask && !isError && (toolName === 'CheckItem' || toolName === 'CreateChecklist')) {
try {
// 計画5: checklist tool の書込先runtimeDir 解決)と同じ root を読む。
const checklistDir = join(runtimeDir ?? join(workspacePath, 'logs'), 'checklists');
if (existsSync(checklistDir)) {
const files = readdirSync(checklistDir).filter(f => f.endsWith('.json'));
if (files.length > 0) {
let latestFile = files[0]!;
let latestMtime = 0;
for (const file of files) {
try {
const { mtimeMs } = statSync(join(checklistDir, file));
if (mtimeMs > latestMtime) {
latestMtime = mtimeMs;
latestFile = file;
}
} catch { /* skip */ }
}
const data = JSON.parse(readFileSync(join(checklistDir, latestFile), 'utf-8'));
const progressBody = JSON.stringify({
type: 'checklist',
name: data.name,
items: data.items,
summary: data.summary,
});
this.repo.addLocalTaskComment(localTaskId!, 'agent', progressBody, 'progress')
.catch(err => logger.warn(`[worker:${this.workerId}] checklist progress comment failed: ${err}`));
}
}
} catch (err) {
logger.warn(`[worker:${this.workerId}] checklist read failed: ${err}`);
}
}
},
};
}
private async handleReflectionJob(job: Job): Promise<void> {
const { runReflectionJob } = await import('./engine/reflection/reflection-runner.js');
try {
// Gateway mode routes by role: send the reflection tier as the key
// (job.requiredRole is 'reflection'), not the worker's model name.
const reflectionRoutingKey = llmRoutingKey({
isGateway: this.getWorkerDef().proxy === true,
role: job.requiredRole,
resolveDirectModel: () => this.model,
roleFallback: 'reflection',
});
const outcome = await runReflectionJob(
{
repo: this.repo,
config: this.config,
llmEndpoint: this.endpoint,
llmModel: reflectionRoutingKey,
// Same credential as normal task LLM calls — a key-enforcing
// gateway 401s reflection without it.
llmApiKey: this.getWorkerDef().apiKey,
llmProxy: this.getWorkerDef().proxy === true,
llmContextLimitTokens: this.contextLimitTokens,
},
job
);
await this.repo.updateJob(job.id, {
status: outcome === 'failed' ? 'failed' : 'succeeded',
currentActivity: null,
});
} catch (e) {
logger.error(`[reflection] runner threw job=${job.id} err=${String(e)}`);
await this.repo.updateJob(job.id, { status: 'failed', currentActivity: null });
}
}
private async handlePieceResult(
result: PieceRunResult,
job: Job,
reporter: LocalProgressReporter,
workspacePath: string,
isLocalTask: boolean,
isSubTask: boolean,
parentJobId: string | null,
): Promise<void> {
const { repo: repoName, issueNumber, id: jobId } = job;
const localTaskId = getLocalTaskId(repoName);
if (result.status === 'completed') {
if (isLocalTask) {
await this.commitLocalWorkspace(issueNumber, workspacePath);
}
await this.repo.updateJob(jobId, { status: 'succeeded', currentActivity: null });
this.enqueuePush(job, 'succeeded');
await maybeEnqueueReflection(this.repo, job, 'succeeded', this.config.reflection, this.config.provider.workers);
let resultBody = result.finalOutput;
if (resultBody) {
resultBody = ensureKeepaGraphs(resultBody);
}
await reporter.reportFinalResult('completed', resultBody);
} else if (result.status === 'waiting_human') {
if (isSubTask && parentJobId) {
await this.repo.updateJob(jobId, { status: 'waiting_human', resumeMovement: result.resumeMovement ?? null, askCount: job.askCount + 1 });
// Sub-task ASK is auto-answered below, so we don't notify on it.
reporter.reportToolUse('ASK', { question: result.finalOutput });
await this.repo.addAuditLog(jobId, 'job_ask_subtask', 'worker', { question: result.finalOutput, resumeMovement: result.resumeMovement });
try {
const answer = await this.answerSubtaskAsk(job, parentJobId, result.finalOutput);
logger.info(`[worker:${this.workerId}] answered subtask ASK for job ${jobId}: ${answer.slice(0, 100)}`);
const newJob = await this.repo.createJob({
repo: repoName,
issueNumber,
instruction: answer,
pieceName: job.pieceName,
askCount: job.askCount + 1,
resumeMovement: result.resumeMovement,
parentJobId: job.parentJobId,
subtaskDepth: job.subtaskDepth,
maxAttempts: 2,
role: job.requiredRole,
// ASK 再投入でも所有権・可視性・スペース・ブラウザセッションのバインドを維持する。
...inheritJobContext(job),
});
await this.repo.updateJob(newJob.id, { worktreePath: workspacePath });
await this.repo.addAuditLog(newJob.id, 'job_queued_subtask_ask_answer', 'worker', { originalJobId: jobId, question: result.finalOutput });
} catch (askErr) {
logger.warn(`[worker:${this.workerId}] failed to answer subtask ASK, leaving as waiting_human: ${askErr}`);
}
} else if (result.waitReason === 'tool_request') {
// Tool-approval pause. MUST persist wait_reason='tool_request' so the
// decide API's resumeToolRequestJob can re-queue it — otherwise the job
// is stuck forever. This is NOT an ASK: it does not consume the per-job
// ASK budget (askCount) and is not reported as a question to answer.
await this.repo.updateJob(jobId, {
status: 'waiting_human',
resumeMovement: result.resumeMovement ?? null,
waitReason: 'tool_request',
});
this.enqueuePush(job, 'waiting_human');
await this.repo.addAuditLog(jobId, 'job_tool_request', 'worker', {
resumeMovement: result.resumeMovement,
});
} else if (result.waitReason === 'package_request') {
// Package-approval pause (RequestPackage). MUST persist
// wait_reason='package_request' so the decide API's
// resumePackageRequestJob can re-queue it — otherwise the job is stuck
// forever. Like tool_request, this is NOT an ASK: it does not consume
// the per-job ASK budget and is not reported as a question.
await this.repo.updateJob(jobId, {
status: 'waiting_human',
resumeMovement: result.resumeMovement ?? null,
waitReason: 'package_request',
});
this.enqueuePush(job, 'waiting_human');
await this.repo.addAuditLog(jobId, 'job_package_request', 'worker', {
resumeMovement: result.resumeMovement,
});
} else {
await this.repo.updateJob(jobId, {
status: 'waiting_human',
resumeMovement: result.resumeMovement ?? null,
askCount: job.askCount + 1,
});
this.enqueuePush(job, 'waiting_human');
await reporter.reportAsk(result.finalOutput);
await this.repo.addAuditLog(jobId, 'job_ask', 'worker', {
question: result.finalOutput,
resumeMovement: result.resumeMovement,
});
}
} else if (result.status === 'waiting_subtasks') {
const subJobs = await this.repo.getSubJobs(jobId);
if (subJobs.length === 0) {
if (result.resumeMovement) {
logger.warn(`[worker:${this.workerId}] job ${jobId} waiting_subtasks but no sub-jobs exist, re-queuing to ${result.resumeMovement}`);
await this.repo.updateJob(jobId, {
status: 'queued',
resumeMovement: result.resumeMovement,
});
} else {
logger.error(`[worker:${this.workerId}] job ${jobId} waiting_subtasks with no sub-jobs and no resumeMovement, failing`);
await this.repo.updateJob(jobId, { status: 'failed' });
}
await this.repo.addAuditLog(jobId, 'job_requeued_no_subtasks', 'worker', {
resumeMovement: result.resumeMovement,
action: result.resumeMovement ? 'requeued' : 'failed',
});
} else {
await this.repo.updateJob(jobId, {
status: 'waiting_subtasks',
resumeMovement: result.resumeMovement ?? null,
});
// P1-3 race guard: a child may have reached its terminal state between the
// park decision (in agent-loop/piece-runner) and this status write. The
// per-child requeue fires on child completion but no-ops while the parent
// is not yet 'waiting_subtasks', so without this re-check the parent could
// park forever. requeueParentJobIfAllSubtasksDone only flips us back to
// 'queued' when every child is already terminal.
const requeuedAtPark = await this.repo.requeueParentJobIfAllSubtasksDone(jobId);
if (requeuedAtPark) {
logger.info(`[worker:${this.workerId}] job ${jobId} all sub-tasks already done at park time, re-queued immediately`);
} else {
await reporter.reportMovementStart('サブタスク待機中...');
}
await this.repo.addAuditLog(jobId, 'job_waiting_subtasks', 'worker', {
resumeMovement: result.resumeMovement,
requeuedAtPark,
});
}
} else if (result.status === 'cancelled') {
logger.info(`[worker:${this.workerId}] job ${jobId} cancelled${result.abortReason ? ` (${result.abortReason})` : ''}`);
// Persist the terminal state + reason. For a USER cancel the row is already
// 'cancelled' (requestJobCancel set it up front) so this is an idempotent
// re-write that additionally records abort_reason. For a DEADLINE timeout
// NOTHING has written the row yet — the guard only aborts the signal — so
// without this the job would stay 'running' in the DB until the stuck-job
// watchdog eventually reaps it. Writing it here closes bug ②'s clean path.
await this.repo.updateJob(jobId, {
status: 'cancelled',
abortReason: result.abortReason ?? ABORT_CODE_CANCELLED,
currentActivity: null,
});
// P1-4 orphan prevention: a cancelled parent never resumes to consume its
// children. Cancel any still-running ones so they don't keep burning GPU.
await this.cancelPendingChildren(jobId, 'parent cancelled');
await reporter.reportFinalResult('cancelled', result.finalOutput);
} else {
// P1-4 orphan prevention: a terminal-failure parent (aborted/error/failed)
// will not consume its children; on retry it re-runs from scratch and
// re-spawns. Either way the current attempt's pending children are orphans —
// cancel them. (ASK / waiting_human is handled above and intentionally keeps
// children: that parent resumes on the user's answer.)
await this.cancelPendingChildren(jobId, `parent ${result.status}`);
const retryDisposition = await this.scheduleRetryOrFail(job, result.finalOutput, workspacePath, result.abortReason ?? null);
if (retryDisposition !== 'requeued_unhealthy') {
await reporter.reportFinalResult(result.status, result.finalOutput);
}
if (retryDisposition === 'failed') {
const outcome = result.status === 'aborted' ? 'aborted' : 'failed';
await maybeEnqueueReflection(this.repo, job, outcome, this.config.reflection, this.config.provider.workers);
}
}
// サブタスク完了時(終端ステータスのみ): 全兄弟ジョブが完了なら親ジョブを再キュー
const SUBTASK_TERMINAL = ['completed', 'error', 'aborted', 'cancelled'];
if (isSubTask && parentJobId && SUBTASK_TERMINAL.includes(result.status)) {
try {
const parentJob = await this.repo.getJob(parentJobId);
if (parentJob?.worktreePath) {
const resultDir = join(parentJob.worktreePath, 'subtasks', String(issueNumber));
mkdirSync(resultDir, { recursive: true });
writeFileSync(
join(resultDir, 'result.md'),
`# サブタスク #${issueNumber} 結果\n\nステータス: ${result.status}\n\n${result.finalOutput}\n`,
'utf-8',
);
}
const requeued = await this.repo.requeueParentJobIfAllSubtasksDone(parentJobId);
if (requeued) {
logger.info(`[worker:${this.workerId}] all sub-tasks done, re-queued parent job ${parentJobId}`);
}
} catch (subErr) {
logger.warn(`[worker:${this.workerId}] sub-task parent re-queue error: ${subErr}`);
}
}
await this.repo.addAuditLog(jobId, `job_${result.status}`, 'worker', {
movementCount: result.movementHistory.length,
abortReason: result.abortReason ?? null,
contextActionCount: result.contextActions.length,
latestContextAction: result.contextActions[result.contextActions.length - 1] ?? null,
});
}
/**
* Orphan prevention (P1-4): cancel this job's not-yet-terminal children when the
* parent ends without consuming them (terminal failure / cancel / retry). Best
* effort — failures are logged, never thrown, so they can't mask the real result.
* Scope: DIRECT children only. A running child observes the cancel via its own
* cancelCheck and aborts. A child that is itself parked (waiting_subtasks) is only
* marked 'cancelled' in the DB and never runs again, so ITS grandchildren keep
* running to completion — wasted GPU, but no hang (nothing waits on them).
* Bounded by maxDepth (default 2); recursive reaping is future scope.
*/
private async cancelPendingChildren(jobId: string, reason: string): Promise<void> {
try {
const subs = await this.repo.getSubJobs(jobId);
let cancelled = 0;
for (const sub of subs) {
if (!SUBTASK_DB_TERMINAL.includes(sub.status)) {
await this.repo.updateJob(sub.id, { status: 'cancelled' });
cancelled += 1;
}
}
if (cancelled > 0) {
logger.info(`[worker:${this.workerId}] cancelled ${cancelled} pending subtask(s) of job ${jobId} (${reason})`);
}
} catch (err) {
logger.warn(`[worker:${this.workerId}] cancelPendingChildren error for job ${jobId}: ${err}`);
}
}
/**
* ②B terminal write for a hard-killed job. The piece run never returned, so
* handlePieceResult won't run — do the minimum here: mark the job terminal
* (with the deadline vs cancel reason so ①'s distinction is preserved), cancel
* still-running children, surface a result comment, and release a parked
* parent when this run was itself a subtask.
*/
private async finalizeHardkill(
job: Job,
reporter: LocalProgressReporter,
isSubTask: boolean,
parentJobId: string | null,
deadline: boolean,
): Promise<void> {
const jobId = job.id;
const abortReason = deadline ? ABORT_CODE_DEADLINE : ABORT_CODE_CANCELLED;
const message = deadline
? '実行時間の上限に達したため、ジョブを自動的に終了しました。(応答しない処理があったため強制終了しています)'
: 'ジョブをキャンセルしました。(応答しない処理があったため強制終了しています)';
await this.repo.updateJob(jobId, { status: 'cancelled', abortReason, currentActivity: null });
await this.cancelPendingChildren(jobId, deadline ? 'parent deadline hardkill' : 'parent cancel hardkill');
await reporter.reportFinalResult('cancelled', message);
await this.repo.addAuditLog(jobId, 'job_hardkill', 'worker', { deadline, abortReason });
if (isSubTask && parentJobId) {
try {
const requeued = await this.repo.requeueParentJobIfAllSubtasksDone(parentJobId);
if (requeued) {
logger.info(`[worker:${this.workerId}] parent job ${parentJobId} re-queued after subtask ${jobId} hardkill`);
}
} catch (err) {
logger.warn(`[worker:${this.workerId}] subtask hardkill parent requeue error: ${err}`);
}
}
}
private resolveModel(piece: PieceDef): string | undefined {
if (piece.model) {
if (this.availableModels.size === 0 || this.availableModels.has(piece.model)) {
return piece.model;
}
logger.warn(`[worker:${this.workerId}] piece model "${piece.model}" not available, falling back to ${this.model ?? '<none>'}`);
}
// If the configured model is not in available models, auto-select the first available one
if (this.model && this.availableModels.size > 0 && !this.availableModels.has(this.model)) {
const autoModel = [...this.availableModels][0]!;
logger.info(`[worker:${this.workerId}] configured model "${this.model}" not available, auto-selecting "${autoModel}"`);
return autoModel;
}
return this.model;
}
private async scheduleRetryOrFail(
job: Job,
errorMsg: string,
workspacePath?: string,
abortReason: string | null = null,
): Promise<'requeued_unhealthy' | 'retry' | 'failed'> {
const { id: jobId, attempt, maxAttempts } = job;
const isLlmConnectionFatal = /connection error:\s*fetch failed|econnrefused|enotfound|etimedout|network error/i.test(errorMsg);
if (isLlmConnectionFatal) {
this.healthy = false;
this.lastHealthError = errorMsg;
this.availableModels.clear();
await this.repo.updateWorkerNodeHealth(this.workerId, {
healthy: false,
lastError: errorMsg,
inflightJobs: this.inflight,
availableModels: [],
});
await this.repo.updateJob(jobId, {
status: 'queued',
workerId: null,
errorSummary: errorMsg,
abortReason,
nextRetryAt: null,
});
writeRetryHandoffSummary({
workspacePath: workspacePath ?? job.worktreePath,
job,
errorMsg,
nextRetryAt: null,
disposition: 'requeued_unhealthy',
});
logger.warn(`[worker:${this.workerId}] job ${jobId} requeued after LLM connection error; worker marked unhealthy`);
return 'requeued_unhealthy';
}
// Decide whether a re-run is worth it.
// - Hard-terminal agent aborts: never retry (deterministic dead end).
// - Recoverable-once aborts (max_iterations / context_overflow): allow ONE
// recovery retry, but give up the moment the same reason repeats —
// `job.abortReason` holds the PREVIOUS attempt's reason, so an equal value
// means the last retry already tried and hit the same wall.
// - Everything else (transient: llm_error / null): retry up to maxAttempts.
const isHardTerminal = abortReason != null && HARD_TERMINAL_ABORT_REASONS.has(abortReason);
const isRecoverableOnce = abortReason != null && RECOVERABLE_ONCE_ABORT_REASONS.has(abortReason);
const sameReasonRepeat = isRecoverableOnce && job.abortReason === abortReason;
const isTerminal = isHardTerminal || sameReasonRepeat;
if (!isTerminal && attempt < maxAttempts) {
const backoffIndex = Math.min(attempt - 1, this.config.retry.backoffSeconds.length - 1);
const backoffSec = this.config.retry.backoffSeconds[backoffIndex] ?? this.config.retry.backoffSeconds[this.config.retry.backoffSeconds.length - 1] ?? 60;
const nextRetryAt = new Date(Date.now() + backoffSec * 1000).toISOString();
await this.repo.updateJob(jobId, {
status: 'retry',
attempt: attempt + 1,
nextRetryAt,
errorSummary: errorMsg,
abortReason,
});
writeRetryHandoffSummary({
workspacePath: workspacePath ?? job.worktreePath,
job,
errorMsg,
nextRetryAt,
disposition: 'retry',
});
logger.info(`[worker:${this.workerId}] job ${jobId} scheduled for retry ${attempt + 1}/${maxAttempts} at ${nextRetryAt}`);
return 'retry';
} else {
await this.repo.updateJob(jobId, { status: 'failed', errorSummary: errorMsg, abortReason });
// V2 push: only on terminal fail. Intermediate retry attempts are
// silenced (matches V1's 4-second debounce intent).
this.enqueuePush(job, 'failed');
writeRetryHandoffSummary({
workspacePath: workspacePath ?? job.worktreePath,
job,
errorMsg,
nextRetryAt: null,
disposition: 'failed',
});
logger.info(`[worker:${this.workerId}] job ${jobId} failed permanently after ${maxAttempts} attempts`);
return 'failed';
}
}
private listDir(dirPath: string): string[] {
try {
return readdirSync(dirPath);
} catch {
return [];
}
}
private async commitLocalWorkspace(
taskId: number,
workspacePath: string,
commitMessage?: string,
): Promise<void> {
const result = await commitWorkspaceChanges({
workspacePath,
branchName: 'main',
commitMessage: commitMessage?.trim() || `agent: update task #${taskId}`,
ignoreEntries: ['input/', 'logs/'],
});
if (!result.changed) {
logger.info(`[worker:${this.workerId}] no local changes to commit for task #${taskId}`);
return;
}
if (result.committed) {
logger.info(`[worker:${this.workerId}] committed local workspace changes for task #${taskId}`);
}
if (result.pushed) {
logger.info(`[worker:${this.workerId}] pushed local workspace changes for task #${taskId}`);
}
}
}