Skip to content

Commit 771c1fc

Browse files
authored
Merge pull request #83 from luohaha/codex/pr-reconciliation-recovery
Improve PR reconciliation recovery and diagnostics
2 parents 78f7c88 + b2231b4 commit 771c1fc

9 files changed

Lines changed: 489 additions & 9 deletions

docs/agent-runners.md

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -169,6 +169,8 @@ By default, Agent Manager polls registered PRs whose last stored state is Draft
169169

170170
Set `pullRequestReconcileIntervalSeconds` in the workspace configuration or dashboard to change the interval dynamically; `0` disables polling. The compatible `--pr-reconcile-interval SECONDS` launch override is also available. Reconciliation requires the launching user to be authenticated with `gh auth login`.
171171

172+
Every `gh` subprocess has a 30-second elapsed timeout and is force-terminated if it hangs. Inspection and snapshot-processing failures are isolated to the affected PR, so the reconciler continues through the other eligible PRs, clears the active polling attempt, and retries failed non-terminal PRs on the next interval. Each failure is logged with `reconciliationStage`, `pullRequestId`, `repository`, `number`, and the nested error details. Diagnostics omit raw `gh` JSON output (which can contain review bodies), process credentials, and recognized credentials in `gh` error output.
173+
172174
## 6. Recovery and failure
173175

174176
- A native session ID is stored as soon as the CLI reports it.

docs/architecture.en.md

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -181,6 +181,8 @@ New review activity is appended as a Reviewer message. PR status changes, CI fai
181181

182182
Observation baselines and trigger-scoped receipts are persisted in SQLite. This prevents duplicate delivery across polling cycles and Agent Manager restarts. Legacy receipts from the former combined `github.pull-request` trigger are copied into the matching split trigger scope during migration. When an older PR is first adopted, existing comments and CI results form the baseline instead of being replayed, while a stale stored PR status is corrected immediately. A conflict is keyed by head SHA, so an unchanged conflict does not repeat while a newly pushed conflicting revision can wake RD again. `pullRequestReconcileIntervalSeconds` changes the interval dynamically; `0` disables polling. The compatible `--pr-reconcile-interval SECONDS` option overrides the file for the launched process only.
183183

184+
Each `gh` subprocess is bounded by a 30-second timeout and force-terminated when that deadline expires. A failed inspection or snapshot reconciliation is recorded with its phase, PR ID, repository, number, and nested error chain; raw GitHub response payloads and process credentials are not logged. Failures are isolated per PR, successful snapshots in the same poll are still applied, and the reconciler clears its in-flight marker after the aggregate attempt so eligible failures are retried on the next configured interval.
185+
184186
GitHub and the PR reconciler exclusively advance PR lifecycle state. The RD Agent uses `code-factory-cli pr register` after creating a PR and may run it again when its own push or edit changes the head SHA, title, or branches, but the underlying Agent API cannot change `draft/open/closed/merged` for an existing PR. Reconciler status messages explicitly say that the state is already persisted, so the RD Agent must not mirror the event.
185187

186188
## 6. State Machines

docs/running-code-factory.md

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -133,6 +133,8 @@ Workspace data is stored outside the managed repository by default:
133133

134134
The foreground CLI prints a startup banner containing the workspace, configuration, database, log path, dashboard URL, API URL, and PR reconciliation interval. Daemon commands print supervisor and manager PIDs plus the daemon log path. Operational logs are structured JSONL and omit prompts, conversation bodies, and raw Agent output. Daemon state and log files use mode `0600`.
135135

136+
PR reconciliation failures are logged individually with `reconciliationStage`, `pullRequestId`, `repository`, `number`, and recursively serialized error or `AggregateError` details. Each local `gh` command is force-terminated after 30 seconds if it hangs; the affected non-terminal PR is retried at the next configured reconciliation interval while other eligible PRs continue. These diagnostics omit raw `gh` JSON output, including review bodies, and do not include process credentials.
137+
136138
The daemon log is append-only diagnostic history: its presence does not mean that a daemon is running and never blocks a later start. `daemon.lock` is also only PID metadata, so a stale copy does not block startup. Live-process detection uses `daemon.json` together with the recorded supervisor PID; an operating-system-released SQLite lock in `daemon.guard.sqlite` serializes supervisor ownership and is safe to leave on disk.
137139

138140
Follow the active logs with:

packages/agent-manager/src/github-client.ts

Lines changed: 52 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,8 @@ import { execFile } from 'node:child_process';
22

33
import type { PullRequest, PullRequestStatus } from './types.js';
44

5+
export const DEFAULT_GH_CLI_TIMEOUT_MS = 30_000;
6+
57
export type GitHubReviewActivityKind = 'comment' | 'review' | 'review_comment';
68
export type GitHubPullRequestMergeability = 'CONFLICTING' | 'MERGEABLE' | 'UNKNOWN';
79

@@ -44,6 +46,11 @@ export interface GitHubClient {
4446
inspectPullRequest(pullRequest: PullRequest): Promise<GitHubPullRequestSnapshot>;
4547
}
4648

49+
export interface GhCliGitHubClientOptions {
50+
executable?: string;
51+
timeoutMs?: number;
52+
}
53+
4754
type JsonObject = Record<string, unknown>;
4855

4956
function objectValue(value: unknown): JsonObject | null {
@@ -141,9 +148,17 @@ function repositoryApiTarget(repository: string): { hostname: string | null; pat
141148

142149
export class GhCliGitHubClient implements GitHubClient {
143150
readonly #workspaceRoot: string;
151+
readonly #executable: string;
152+
readonly #timeoutMs: number;
144153

145-
constructor(workspaceRoot: string) {
154+
constructor(workspaceRoot: string, options: GhCliGitHubClientOptions = {}) {
155+
const timeoutMs = options.timeoutMs ?? DEFAULT_GH_CLI_TIMEOUT_MS;
156+
if (!Number.isSafeInteger(timeoutMs) || timeoutMs < 1) {
157+
throw new RangeError('GitHub CLI timeout must be a positive integer');
158+
}
146159
this.#workspaceRoot = workspaceRoot;
160+
this.#executable = options.executable ?? 'gh';
161+
this.#timeoutMs = timeoutMs;
147162
}
148163

149164
async inspectPullRequest(pullRequest: PullRequest): Promise<GitHubPullRequestSnapshot> {
@@ -193,23 +208,54 @@ export class GhCliGitHubClient implements GitHubClient {
193208

194209
private runJson(args: string[]): Promise<unknown> {
195210
return new Promise((resolve, reject) => {
196-
execFile('gh', args, {
211+
const command = ghCommandLabel(args);
212+
execFile(this.#executable, args, {
197213
cwd: this.#workspaceRoot,
198214
env: process.env,
199215
encoding: 'utf8',
200216
maxBuffer: 10 * 1024 * 1024,
217+
timeout: this.#timeoutMs,
218+
killSignal: 'SIGKILL',
201219
}, (error, stdout, stderr) => {
202220
if (error) {
203-
const detail = String(stderr).trim() || error.message;
204-
reject(new Error(`gh ${args.slice(0, 2).join(' ')} failed: ${detail}`));
221+
if (error.killed && error.signal === 'SIGKILL') {
222+
const timeoutError = new Error(`${command} timed out after ${this.#timeoutMs}ms`);
223+
timeoutError.name = 'TimeoutError';
224+
reject(timeoutError);
225+
return;
226+
}
227+
reject(new Error(`${command} failed: ${ghFailureDetail(error, String(stderr))}`));
205228
return;
206229
}
207230
try {
208231
resolve(JSON.parse(String(stdout)) as unknown);
209-
} catch (parseError) {
210-
reject(new Error(`gh returned invalid JSON: ${parseError instanceof Error ? parseError.message : String(parseError)}`));
232+
} catch {
233+
// stdout can contain review bodies. Identify the parse failure without
234+
// copying the response payload or the engine's source excerpt to logs.
235+
reject(new SyntaxError(`${command} returned invalid JSON`));
211236
}
212237
});
213238
});
214239
}
215240
}
241+
242+
function ghCommandLabel(args: readonly string[]): string {
243+
return args[0] === 'pr' && args[1] === 'view' ? 'gh pr view' : `gh ${args[0] ?? 'command'}`;
244+
}
245+
246+
function ghFailureDetail(error: Error & { code?: string | number | null; signal?: NodeJS.Signals | null }, stderr: string): string {
247+
const diagnostic = redactCredentials(stderr.trim()).slice(0, 4_000);
248+
if (diagnostic) return diagnostic;
249+
if (error.code !== undefined && error.code !== null) {
250+
return typeof error.code === 'number' ? `exit code ${error.code}` : `process error ${error.code}`;
251+
}
252+
if (error.signal) return `terminated by ${error.signal}`;
253+
return error.name;
254+
}
255+
256+
function redactCredentials(value: string): string {
257+
return value
258+
.replace(/\b(?:gh[pousr]_[A-Za-z0-9_]{16,}|github_pat_[A-Za-z0-9_]{16,})\b/g, '[REDACTED]')
259+
.replace(/(authorization:\s*(?:bearer|token)\s+)[^\s]+/gi, '$1[REDACTED]')
260+
.replace(/(https?:\/\/)[^\s/@]+@/gi, '$1[REDACTED]@');
261+
}

packages/agent-manager/src/logger.ts

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -215,6 +215,7 @@ function safeStringify(value: unknown): string {
215215
message: current.message,
216216
...(current.stack ? { stack: current.stack } : {}),
217217
...(current.cause === undefined ? {} : { cause: current.cause }),
218+
...(current instanceof AggregateError ? { errors: current.errors } : {}),
218219
};
219220
}
220221
return current;

packages/agent-manager/src/pull-request-reconciler.ts

Lines changed: 28 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -110,17 +110,42 @@ export class PullRequestReconciler {
110110
const errors: Error[] = [];
111111
for (const pullRequest of this.#store.listPullRequests()
112112
.filter((item) => item.status === 'draft' || item.status === 'open')) {
113+
let snapshot: Awaited<ReturnType<GitHubClient['inspectPullRequest']>>;
114+
try {
115+
snapshot = await this.#githubClient.inspectPullRequest(pullRequest);
116+
} catch (error) {
117+
errors.push(this.recordFailure(pullRequest, 'inspection', error));
118+
continue;
119+
}
120+
if (this.#isClosed()) return;
113121
try {
114-
const snapshot = await this.#githubClient.inspectPullRequest(pullRequest);
115-
if (this.#isClosed()) return;
116122
this.reconcileSnapshot(registrations, pullRequest, snapshot);
117123
} catch (error) {
118-
errors.push(error instanceof Error ? error : new Error(String(error)));
124+
errors.push(this.recordFailure(pullRequest, 'snapshot', error));
119125
}
120126
}
121127
if (errors.length > 0) throw new AggregateError(errors, `${errors.length} pull request(s) could not be reconciled`);
122128
}
123129

130+
private recordFailure(
131+
pullRequest: PullRequest,
132+
reconciliationStage: 'inspection' | 'snapshot',
133+
error: unknown,
134+
): Error {
135+
const underlyingError = error instanceof Error ? error : new Error(String(error));
136+
this.#logger.error('Pull request reconciliation failed', {
137+
reconciliationStage,
138+
pullRequestId: pullRequest.id,
139+
repository: pullRequest.repository,
140+
number: pullRequest.number,
141+
error: underlyingError,
142+
});
143+
return new Error(
144+
`${reconciliationStage} failed for ${pullRequest.repository}#${pullRequest.number} (${pullRequest.id})`,
145+
{ cause: underlyingError },
146+
);
147+
}
148+
124149
private reconcileSnapshot(
125150
registrations: readonly PullRequestTriggerRegistration[],
126151
pullRequest: PullRequest,
Lines changed: 130 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,130 @@
1+
import assert from 'node:assert/strict';
2+
import { execFileSync } from 'node:child_process';
3+
import { existsSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs';
4+
import { tmpdir } from 'node:os';
5+
import { join } from 'node:path';
6+
import test from 'node:test';
7+
8+
import { GhCliGitHubClient } from '../src/github-client.ts';
9+
import type { PullRequest } from '../src/types.ts';
10+
11+
const pullRequest: PullRequest = {
12+
id: 'pr-timeout',
13+
requirementId: 'req-timeout',
14+
repository: 'acme/widgets',
15+
number: 81,
16+
url: 'https://github.com/acme/widgets/pull/81',
17+
title: 'Timeout recovery',
18+
baseBranch: 'main',
19+
headBranch: 'timeout-recovery',
20+
headSha: 'abc123',
21+
status: 'open',
22+
createdAt: '2026-09-21T00:00:00.000Z',
23+
updatedAt: '2026-09-21T00:00:00.000Z',
24+
};
25+
26+
test('GhCliGitHubClient kills a hung gh subprocess at its timeout', async () => {
27+
const directory = mkdtempSync(join(tmpdir(), 'code-factory-gh-timeout-'));
28+
const executable = join(directory, 'gh');
29+
const pidFile = join(directory, 'pids');
30+
const hangTarget = join(directory, 'hung-gh-marker');
31+
writeFileSync(hangTarget, '');
32+
writeFileSync(executable, [
33+
'#!/bin/sh',
34+
'if [ "$1" = "api" ]; then printf "[]"; exit 0; fi',
35+
`printf '%s\\n' "$$" >> '${pidFile.replaceAll("'", "'\\''")}'`,
36+
`exec tail -f '${hangTarget.replaceAll("'", "'\\''")}'`,
37+
].join('\n'), { mode: 0o700 });
38+
39+
try {
40+
const client = new GhCliGitHubClient(directory, { executable, timeoutMs: 2_000 });
41+
const startedAt = Date.now();
42+
await assert.rejects(client.inspectPullRequest(pullRequest), (error: unknown) => {
43+
assert.ok(error instanceof Error);
44+
assert.equal(error.name, 'TimeoutError');
45+
assert.match(error.message, /^gh (?:pr view|api) timed out after 2000ms$/);
46+
return true;
47+
});
48+
assert.ok(Date.now() - startedAt < 5_000, 'the hung command should reject promptly');
49+
50+
assert.ok(existsSync(pidFile), 'the fake gh command should reach its hung state before the timeout');
51+
const pids = readPids(pidFile);
52+
assert.equal(pids.length, 1);
53+
await waitFor(() => pids.every((pid) => !markedProcessIsRunning(pid, hangTarget)));
54+
assert.ok(pids.every((pid) => !markedProcessIsRunning(pid, hangTarget)), 'the hung gh subprocess should not be running');
55+
} finally {
56+
for (const pid of existsSync(pidFile) ? readPids(pidFile) : []) {
57+
if (markedProcessIsRunning(pid, hangTarget)) process.kill(pid, 'SIGKILL');
58+
}
59+
rmSync(directory, { recursive: true, force: true });
60+
}
61+
});
62+
63+
test('GhCliGitHubClient reports parse failures without copying gh stdout into the error', async () => {
64+
const directory = mkdtempSync(join(tmpdir(), 'code-factory-gh-parse-'));
65+
const executable = join(directory, 'gh');
66+
writeFileSync(executable, [
67+
'#!/usr/bin/env node',
68+
"process.stdout.write('PRIVATE REVIEW BODY WITH malformed JSON');",
69+
].join('\n'), { mode: 0o700 });
70+
71+
try {
72+
const client = new GhCliGitHubClient(directory, { executable, timeoutMs: 2_000 });
73+
await assert.rejects(client.inspectPullRequest(pullRequest), (error: unknown) => {
74+
assert.ok(error instanceof SyntaxError);
75+
assert.match(error.message, /^gh (?:pr view|api) returned invalid JSON$/);
76+
assert.doesNotMatch(error.message, /PRIVATE REVIEW BODY/);
77+
return true;
78+
});
79+
} finally {
80+
rmSync(directory, { recursive: true, force: true });
81+
}
82+
});
83+
84+
test('GhCliGitHubClient redacts credentials from gh stderr diagnostics', async () => {
85+
const directory = mkdtempSync(join(tmpdir(), 'code-factory-gh-redaction-'));
86+
const executable = join(directory, 'gh');
87+
const token = 'ghp_abcdefghijklmnopqrstuvwxyz123456';
88+
writeFileSync(executable, [
89+
'#!/usr/bin/env node',
90+
`process.stderr.write(${JSON.stringify(`HTTP 401 Authorization: Bearer ${token} https://user:password@github.com`)});`,
91+
'process.exitCode = 1;',
92+
].join('\n'), { mode: 0o700 });
93+
94+
try {
95+
const client = new GhCliGitHubClient(directory, { executable, timeoutMs: 2_000 });
96+
await assert.rejects(client.inspectPullRequest(pullRequest), (error: unknown) => {
97+
assert.ok(error instanceof Error);
98+
assert.match(error.message, /\[REDACTED\]/);
99+
assert.doesNotMatch(error.message, new RegExp(token));
100+
assert.doesNotMatch(error.message, /user:password/);
101+
return true;
102+
});
103+
} finally {
104+
rmSync(directory, { recursive: true, force: true });
105+
}
106+
});
107+
108+
function readPids(filePath: string): number[] {
109+
return readFileSync(filePath, 'utf8').trim().split('\n').filter(Boolean).map(Number);
110+
}
111+
112+
function markedProcessIsRunning(pid: number, marker: string): boolean {
113+
try {
114+
const output = execFileSync('ps', ['-o', 'stat=', '-o', 'command=', '-p', String(pid)], { encoding: 'utf8' }).trim();
115+
const match = output.match(/^(\S+)\s+(.*)$/s);
116+
if (!match) return false;
117+
const [, state, command] = match;
118+
return !state!.startsWith('Z') && command!.includes(marker);
119+
} catch {
120+
return false;
121+
}
122+
}
123+
124+
async function waitFor(predicate: () => boolean, timeoutMs = 2_000): Promise<void> {
125+
const deadline = Date.now() + timeoutMs;
126+
while (!predicate()) {
127+
if (Date.now() >= deadline) throw new Error(`Condition was not met within ${timeoutMs}ms`);
128+
await new Promise<void>((resolve) => setTimeout(resolve, 10));
129+
}
130+
}

packages/agent-manager/test/logger.test.ts

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -61,6 +61,25 @@ test('child logger adds context and safely serializes errors and circular values
6161
assert.equal(entry.count, '1');
6262
});
6363

64+
test('logger preserves nested AggregateError details and causes', () => {
65+
const stderr = new MemoryWriter();
66+
const logger = createLogger({ level: 'error', stdout: new MemoryWriter(), stderr });
67+
const nested = new AggregateError([
68+
new Error('gh pr view failed'),
69+
new AggregateError([new SyntaxError('gh api returned invalid JSON')], 'inline comments failed'),
70+
], 'pull request inspection failed');
71+
72+
logger.error('reconciliation failed', {
73+
error: new Error('acme/widgets#81 failed', { cause: nested }),
74+
});
75+
76+
const entry = JSON.parse(stderr.lines[0]!) as {
77+
error: { cause: { errors: Array<{ message: string; errors?: Array<{ message: string }> }> } };
78+
};
79+
assert.equal(entry.error.cause.errors[0]?.message, 'gh pr view failed');
80+
assert.equal(entry.error.cause.errors[1]?.errors?.[0]?.message, 'gh api returned invalid JSON');
81+
});
82+
6483
test('file logger appends every log level to a private JSONL file', async () => {
6584
const directory = mkdtempSync(join(tmpdir(), 'code-factory-logger-'));
6685
const filePath = join(directory, 'nested', 'agent-manager.log');

0 commit comments

Comments
 (0)