Skip to content

Fix provider interruption handling and safe run resume #5

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
hyspacex wants to merge 1 commit into
base: main
Choose a base branch
from
+489 −98
Open

Fix provider interruption handling and safe run resume #5

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -169,6 +169,8 @@ npm run harness -- status
npm run harness -- status <run-id>
```

Resume continues an active sprint even when it has reached `maxSprints`. Completed evaluation rounds are reused only after their canonical results, verdicts, and frozen evidence are validated. Incomplete rounds are replayed, and failed rounds retain their repair instructions and generator session.

## Providers

Harness CLI is provider-agnostic. You choose which AI backend builds your app — and you can use different providers for different roles.
Expand Down
53 changes: 45 additions & 8 deletions src/core/harness.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,6 +89,7 @@ interface SprintProgress {
nextRound: number;
latestEvalPath: string | null;
latestEvalParsed: Record<string, unknown> | null;
latestRepairDirectivePath: string | null;
allEvalPaths: string[];
allFrozenEvidenceDirs: string[];
}
Expand Down Expand Up @@ -450,7 +451,7 @@ export class HarnessRunner {
runState.status = 'running';
await this.saveRunState(runState);

while (runState.sprint < this.config.maxSprints) {
while (runState.sprint < this.config.maxSprints || this.getActiveFeature(backlog, runState)) {
const feature = await this.getCurrentOrNextFeature(runState, backlog);
if (!feature) {
const backlogComplete = backlog.features.every((candidate) => candidate.status === 'done');
Expand All @@ -475,7 +476,7 @@ export class HarnessRunner {
const progress = await this.loadSprintProgress(runState, evalCriteria);
let latestEvalPath = progress.latestEvalPath;
let latestEvalParsed = progress.latestEvalParsed;
let latestRepairDirectivePath: string | null = null;
let latestRepairDirectivePath = progress.latestRepairDirectivePath;
const allEvalPaths = [...progress.allEvalPaths];
let latestFrozenEvidenceDir = progress.allFrozenEvidenceDirs.at(-1) || null;
const allFrozenEvidenceDirs = [...progress.allFrozenEvidenceDirs];
Expand Down Expand Up @@ -812,26 +813,62 @@ export class HarnessRunner {
const allFrozenEvidenceDirs: string[] = [];
let latestEvalPath: string | null = null;
let latestEvalParsed: Record<string, unknown> | null = null;
let latestRepairDirectivePath: string | null = null;
let passed = false;

// Only a contiguous sequence of finalized evaluations consumes repair rounds.
// Provider-written output alone may have survived an interrupted task.
runState.currentVerdictPath = null;
for (let round = 0; round <= this.config.maxRepairRounds; round += 1) {
const evalPath = this.evalPath(runState.sprint, round, runState.runDir);
if (!(await fileExists(evalPath))) continue;
const evalJsonPath = this.evalJsonPath(runState.sprint, round, runState.runDir);
const verdictPath = this.verdictPath(runState.sprint, round, runState.runDir);
if (!(await fileExists(evalPath)) || !(await fileExists(evalJsonPath)) || !(await fileExists(verdictPath))) break;

allEvalPaths.push(evalPath);
latestEvalPath = evalPath;
latestEvalParsed = await this.readParsedTaskLog(this.evaluatorLogName(runState.sprint, round), runState);
const verdict = await readJson<HarnessVerdict | null>(verdictPath, null);
if (!verdict || verdict.version !== 1 || verdict.sprint !== runState.sprint
|| verdict.evaluationRound !== round || verdict.featureId !== runState.currentFeatureId) break;
const canonicalEval = await this.readCanonicalEvaluation(evalJsonPath);
if (canonicalEval.sprint !== runState.sprint || canonicalEval.evaluationRound !== round
|| canonicalEval.feature.id !== runState.currentFeatureId) break;

const frozenEvidenceDir = this.frozenEvidenceDir(runState.sprint, round, runState.runDir);
if (await fileExists(frozenEvidenceDir)) {
const hasFrozenEvidence = await fileExists(frozenEvidenceDir);
if ((verdict.reason !== 'smoke_failure' || (await fileExists(this.evidenceDir(runState.sprint, round, runState.runDir))))
&& !(await fileExists(this.frozenEvidenceManifestPath(frozenEvidenceDir)))) break;
if (hasFrozenEvidence) {
await this.assertFrozenEvidenceIntact(runState, frozenEvidenceDir);
allFrozenEvidenceDirs.push(frozenEvidenceDir);
}

allEvalPaths.push(evalPath);
latestEvalPath = evalPath;
latestEvalParsed = canonicalEval as unknown as Record<string, unknown>;
runState.currentEvalPath = evalPath;
runState.currentEvalJsonPath = evalJsonPath;
runState.currentVerdictPath = verdictPath;
passed = verdict.passed && resolvePass(latestEvalParsed, evalCriteria, this.getEffectivePassBarOverrides(runState));
if (passed) break;
// Also recover an interruption between writing the verdict and directive.
latestRepairDirectivePath = await this.writeRepairDirective(
runState, round, verdict, evalCriteria, hasFrozenEvidence ? frozenEvidenceDir : null,
);
}

if (!passed) {
// A replay must not inherit a verdict from an interrupted attempt, including
// later files after a gap in the completed-round sequence.
for (let round = allEvalPaths.length; round <= this.config.maxRepairRounds; round += 1) {
await fs.rm(this.verdictPath(runState.sprint, round, runState.runDir), { force: true });
}
}

return {
passed: resolvePass(latestEvalParsed, evalCriteria, runState.currentNegotiation?.passBarOverrides ?? {}),
passed,
nextRound: allEvalPaths.length,
latestEvalPath,
latestEvalParsed,
latestRepairDirectivePath,
allEvalPaths,
allFrozenEvidenceDirs,
};
Expand Down
142 changes: 74 additions & 68 deletions src/core/providers/claude-sdk.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,74 +36,9 @@ export class ClaudeSdkProvider implements ProviderRuntime {
async runTask(task: TaskDefinition): Promise<TaskResult> {
const options = this.resolveOptionsForTask(task);

let assistantText = '';
let resultText = '';
let structuredOutput: Record<string, unknown> | null = null;
let failureMessage: string | null = null;
let claudeCodeVersion: string | undefined;
const responseModels = new Set<string>();
let sessionId: string | undefined;

try {
// eslint-disable-next-line @typescript-eslint/no-explicit-any
for await (const message of query({ prompt: task.prompt, options } as any)) {
const msg = message as Record<string, unknown>;

if (msg.type === 'system' && msg.subtype === 'init') {
sessionId = msg.session_id as string;
claudeCodeVersion = msg.claude_code_version as string;
}

if (msg.type === 'assistant') {
const assistant = msg.message as Record<string, unknown>;
if (typeof assistant?.model === 'string') responseModels.add(assistant.model);
const content = assistant?.content;
if (Array.isArray(content)) {
for (const block of content) {
const b = block as Record<string, unknown>;
if (b.type === 'text') {
assistantText += b.text as string;
}
if (b.type === 'tool_use') {
this.onUpdate({ sessionUpdate: 'tool_call', title: b.name as string }, task);
}
}
}
}

if (msg.type === 'result' && msg.subtype === 'success') {
resultText = (msg.result as string) || '';
if (isPlainObject(msg.structured_output)) {
structuredOutput = msg.structured_output;
}
}

failureMessage = claudeResultError(msg) || failureMessage;
}
} catch (error) {
if (failureMessage) throw new Error(`Claude Agent SDK: ${failureMessage}`);
// SDK process crashed — use whatever we collected so far.
const partialText = (resultText || assistantText).trim();
if (partialText) {
const parsed = extractJsonObject(partialText);
if (parsed) {
return { rawText: partialText, parsed, meta: { sessionId, crashed: true } };
}
}
throw error;
}

if (failureMessage) {
throw new Error(`Claude Agent SDK: ${failureMessage}`);
}

const rawText = (resultText || assistantText).trim();

return {
rawText,
parsed: structuredOutput || extractJsonObject(rawText),
meta: { sessionId, structuredOutput: !!structuredOutput, claudeCodeVersion, responseModels: [...responseModels] },
};
// eslint-disable-next-line @typescript-eslint/no-explicit-any
return collectClaudeTaskResult(query({ prompt: task.prompt, options } as any),
(update) => this.onUpdate(update, task));
}

/**
Expand Down Expand Up @@ -168,6 +103,77 @@ export class ClaudeSdkProvider implements ProviderRuntime {
}
}

/** Consume the SDK stream; only a terminal successful result completes a task. */
export async function collectClaudeTaskResult(
messages: AsyncIterable<unknown>,
onUpdate: (update: Record<string, unknown>) => void = () => {},
): Promise<TaskResult> {
let assistantText = '';
let resultText = '';
let structuredOutput: Record<string, unknown> | null = null;
let failureMessage: string | null = null;
let completed = false;
let claudeCodeVersion: string | undefined;
const responseModels = new Set<string>();
let sessionId: string | undefined;

try {
for await (const message of messages) {
const msg = message as Record<string, unknown>;

if (msg.type === 'system' && msg.subtype === 'init') {
sessionId = msg.session_id as string;
claudeCodeVersion = msg.claude_code_version as string;
}

if (msg.type === 'assistant') {
const assistant = msg.message as Record<string, unknown>;
if (typeof assistant?.model === 'string') responseModels.add(assistant.model);
const content = assistant?.content;
if (Array.isArray(content)) {
for (const block of content) {
const b = block as Record<string, unknown>;
if (b.type === 'text') {
assistantText += b.text as string;
}
if (b.type === 'tool_use') {
onUpdate({ sessionUpdate: 'tool_call', title: b.name as string });
}
}
}
}

if (msg.type === 'result' && msg.subtype === 'success') {
completed = !msg.is_error;
resultText = (msg.result as string) || '';
if (isPlainObject(msg.structured_output)) {
structuredOutput = msg.structured_output;
}
}

failureMessage = claudeResultError(msg) || failureMessage;
}
} catch (error) {
if (failureMessage) throw new Error(`Claude Agent SDK: ${failureMessage}`);
// Partial assistant JSON is not a successful terminal SDK result.
throw error;
}

if (failureMessage) {
throw new Error(`Claude Agent SDK: ${failureMessage}`);
}

if (!completed) throw new Error('Claude Agent SDK: stream ended without a successful result');

const rawText = (resultText || assistantText).trim();

return {
rawText,
parsed: structuredOutput || extractJsonObject(rawText),
meta: { sessionId, structuredOutput: !!structuredOutput, claudeCodeVersion, responseModels: [...responseModels] },
};
}

// SDK failures have specific subtypes (for example error_max_turns), not just "error".
export function claudeResultError(message: Record<string, unknown>): string | null {
if (message.type !== 'result') return null;
Expand Down
Loading
Loading