fix worker forreal
This commit is contained in:
@@ -82,8 +82,10 @@ const toolNameFromRecord = (record: Record<string, unknown> | null) =>
|
|||||||
record?.toolName ??
|
record?.toolName ??
|
||||||
record?.name ??
|
record?.name ??
|
||||||
record?.function ??
|
record?.function ??
|
||||||
record?.type ??
|
(stringify(record?.type).toLowerCase().includes('exec') ||
|
||||||
record?.command ??
|
record?.command
|
||||||
|
? 'Command'
|
||||||
|
: record?.type) ??
|
||||||
'tool',
|
'tool',
|
||||||
);
|
);
|
||||||
|
|
||||||
@@ -103,9 +105,15 @@ const toolOutputFromRecord = (
|
|||||||
) =>
|
) =>
|
||||||
stringify(
|
stringify(
|
||||||
record?.output ??
|
record?.output ??
|
||||||
|
record?.aggregated_output ??
|
||||||
|
record?.stdout ??
|
||||||
|
record?.stderr ??
|
||||||
record?.result ??
|
record?.result ??
|
||||||
record?.content ??
|
record?.content ??
|
||||||
record?.text ??
|
record?.text ??
|
||||||
|
(record?.exit_code !== undefined
|
||||||
|
? `exit code: ${stringify(record.exit_code)}`
|
||||||
|
: undefined) ??
|
||||||
fallback,
|
fallback,
|
||||||
);
|
);
|
||||||
|
|
||||||
@@ -121,11 +129,17 @@ const recordLooksLikeTool = (
|
|||||||
recordType.includes('tool') ||
|
recordType.includes('tool') ||
|
||||||
recordType.includes('function_call') ||
|
recordType.includes('function_call') ||
|
||||||
recordType.includes('local_shell_call') ||
|
recordType.includes('local_shell_call') ||
|
||||||
|
recordType.includes('exec_command') ||
|
||||||
|
recordType.includes('command') ||
|
||||||
recordType.includes('mcp') ||
|
recordType.includes('mcp') ||
|
||||||
Boolean(record?.tool ?? record?.tool_name ?? record?.name)
|
Boolean(record?.tool ?? record?.tool_name ?? record?.name ?? record?.command)
|
||||||
);
|
);
|
||||||
};
|
};
|
||||||
|
|
||||||
|
const isCodexConfigWarning = (message: string) =>
|
||||||
|
message.includes('`[features].codex_hooks` is deprecated') ||
|
||||||
|
message.includes('Use `[features].hooks` instead');
|
||||||
|
|
||||||
export const normalizeCodexJsonLine = (
|
export const normalizeCodexJsonLine = (
|
||||||
line: string,
|
line: string,
|
||||||
): NormalizedAgentEvent[] => {
|
): NormalizedAgentEvent[] => {
|
||||||
@@ -205,13 +219,22 @@ export const normalizeCodexJsonLine = (
|
|||||||
itemType.includes('message') ||
|
itemType.includes('message') ||
|
||||||
itemType.includes('agent_message'))
|
itemType.includes('agent_message'))
|
||||||
) {
|
) {
|
||||||
events.push({ kind: 'assistant_delta', content: text });
|
events.push({
|
||||||
|
kind: 'assistant_delta',
|
||||||
|
content: itemType.includes('agent_message') ? `${text.trim()}\n\n` : text,
|
||||||
|
externalMessageId: stringify(item?.id ?? event.id),
|
||||||
|
});
|
||||||
}
|
}
|
||||||
const error = event.error ?? item?.error;
|
const error = event.error ?? item?.error;
|
||||||
if (error || itemType === 'error') {
|
if (error || itemType === 'error') {
|
||||||
|
const message = stringify(error ?? item?.message ?? event.message);
|
||||||
|
if (isCodexConfigWarning(message)) {
|
||||||
|
events.push({ kind: 'status', status: message });
|
||||||
|
return events;
|
||||||
|
}
|
||||||
events.push({
|
events.push({
|
||||||
kind: 'error',
|
kind: 'error',
|
||||||
message: stringify(error ?? item?.message ?? event.message),
|
message,
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
const command =
|
const command =
|
||||||
|
|||||||
@@ -126,12 +126,42 @@ export const getDiff = async (
|
|||||||
export const getWorktreeDiff = async (
|
export const getWorktreeDiff = async (
|
||||||
repoDir: string,
|
repoDir: string,
|
||||||
redact: (value: string) => string,
|
redact: (value: string) => string,
|
||||||
) =>
|
) => {
|
||||||
await run('git', ['diff', '--', '.'], {
|
const trackedDiff = await run('git', ['diff', '--', '.'], {
|
||||||
cwd: repoDir,
|
cwd: repoDir,
|
||||||
redact,
|
redact,
|
||||||
timeoutMs: 60_000,
|
timeoutMs: 60_000,
|
||||||
});
|
});
|
||||||
|
const untracked = await run(
|
||||||
|
'git',
|
||||||
|
['ls-files', '--others', '--exclude-standard'],
|
||||||
|
{
|
||||||
|
cwd: repoDir,
|
||||||
|
redact,
|
||||||
|
timeoutMs: 60_000,
|
||||||
|
},
|
||||||
|
);
|
||||||
|
const untrackedDiffs: string[] = [];
|
||||||
|
for (const filePath of untracked.output.split('\n').filter(Boolean)) {
|
||||||
|
const diff = await run(
|
||||||
|
'git',
|
||||||
|
['diff', '--no-index', '--', '/dev/null', filePath],
|
||||||
|
{
|
||||||
|
cwd: repoDir,
|
||||||
|
redact,
|
||||||
|
timeoutMs: 60_000,
|
||||||
|
},
|
||||||
|
);
|
||||||
|
if (diff.output.trim()) untrackedDiffs.push(diff.output);
|
||||||
|
}
|
||||||
|
return {
|
||||||
|
exitCode:
|
||||||
|
trackedDiff.exitCode === 0 && untracked.exitCode === 0 ? 0 : 1,
|
||||||
|
output: [trackedDiff.output, ...untrackedDiffs]
|
||||||
|
.filter((part) => part.trim())
|
||||||
|
.join('\n'),
|
||||||
|
};
|
||||||
|
};
|
||||||
|
|
||||||
export const getStatus = async (
|
export const getStatus = async (
|
||||||
repoDir: string,
|
repoDir: string,
|
||||||
|
|||||||
@@ -54,6 +54,7 @@ export const runInJobContainer = async (args: {
|
|||||||
{
|
{
|
||||||
all: true,
|
all: true,
|
||||||
reject: false,
|
reject: false,
|
||||||
|
stdin: 'ignore',
|
||||||
timeout: args.timeoutMs,
|
timeout: args.timeoutMs,
|
||||||
},
|
},
|
||||||
);
|
);
|
||||||
@@ -102,7 +103,7 @@ export const startWorkspaceContainer = async (args: {
|
|||||||
env.jobImage,
|
env.jobImage,
|
||||||
...(args.command ?? ['sleep', 'infinity']),
|
...(args.command ?? ['sleep', 'infinity']),
|
||||||
],
|
],
|
||||||
{ all: true },
|
{ all: true, stdin: 'ignore' },
|
||||||
);
|
);
|
||||||
return {
|
return {
|
||||||
containerId: result.stdout.trim(),
|
containerId: result.stdout.trim(),
|
||||||
@@ -117,7 +118,7 @@ const getPublishedPort = async (containerName: string, containerPort: number) =>
|
|||||||
const result = await execa(
|
const result = await execa(
|
||||||
containerRuntime(),
|
containerRuntime(),
|
||||||
['port', containerName, `${containerPort}/tcp`],
|
['port', containerName, `${containerPort}/tcp`],
|
||||||
{ all: true, reject: false },
|
{ all: true, reject: false, stdin: 'ignore' },
|
||||||
);
|
);
|
||||||
const output = result.all.trim();
|
const output = result.all.trim();
|
||||||
const match = /:(\d+)\s*$/.exec(output);
|
const match = /:(\d+)\s*$/.exec(output);
|
||||||
@@ -147,6 +148,7 @@ export const execInWorkspaceContainer = async (args: {
|
|||||||
{
|
{
|
||||||
all: true,
|
all: true,
|
||||||
reject: false,
|
reject: false,
|
||||||
|
stdin: 'ignore',
|
||||||
timeout: args.timeoutMs,
|
timeout: args.timeoutMs,
|
||||||
},
|
},
|
||||||
);
|
);
|
||||||
@@ -186,6 +188,7 @@ export const streamInJobContainer = async (args: {
|
|||||||
{
|
{
|
||||||
all: true,
|
all: true,
|
||||||
reject: false,
|
reject: false,
|
||||||
|
stdin: 'ignore',
|
||||||
timeout: args.timeoutMs,
|
timeout: args.timeoutMs,
|
||||||
},
|
},
|
||||||
);
|
);
|
||||||
@@ -220,7 +223,19 @@ export const streamInJobContainer = async (args: {
|
|||||||
consume(chunk, 'stderr', args.onStderrLine),
|
consume(chunk, 'stderr', args.onStderrLine),
|
||||||
);
|
);
|
||||||
});
|
});
|
||||||
const result = await subprocess;
|
let result: Awaited<typeof subprocess>;
|
||||||
|
try {
|
||||||
|
result = await subprocess;
|
||||||
|
} catch (error) {
|
||||||
|
await lineHandlers;
|
||||||
|
const outputText = output.join('');
|
||||||
|
const message =
|
||||||
|
error instanceof Error ? error.message : 'Container command failed.';
|
||||||
|
return {
|
||||||
|
exitCode: 1,
|
||||||
|
output: args.redact(`${outputText}${outputText ? '\n' : ''}${message}`),
|
||||||
|
};
|
||||||
|
}
|
||||||
await lineHandlers;
|
await lineHandlers;
|
||||||
if (stdoutBuffer && args.onStdoutLine) {
|
if (stdoutBuffer && args.onStdoutLine) {
|
||||||
await args.onStdoutLine(args.redact(stdoutBuffer));
|
await args.onStdoutLine(args.redact(stdoutBuffer));
|
||||||
|
|||||||
@@ -110,6 +110,7 @@ type ActiveWorkspace = {
|
|||||||
codexSessionId?: string;
|
codexSessionId?: string;
|
||||||
agentTurnActive?: boolean;
|
agentTurnActive?: boolean;
|
||||||
resolveTurn?: () => void;
|
resolveTurn?: () => void;
|
||||||
|
lastRecordedDiffSignature?: string;
|
||||||
};
|
};
|
||||||
|
|
||||||
type FileTreeNode = {
|
type FileTreeNode = {
|
||||||
@@ -430,8 +431,8 @@ const codexModel = (claim: Claim) => {
|
|||||||
return model.includes('/') ? model.split('/').at(-1) ?? model : model;
|
return model.includes('/') ? model.split('/').at(-1) ?? model : model;
|
||||||
};
|
};
|
||||||
|
|
||||||
const codexModelFlag = (claim: Claim) =>
|
const codexModelArgs = (claim: Claim) =>
|
||||||
isCodexLoginProfile(claim) ? '' : ` --model ${quoteShell(codexModel(claim))}`;
|
isCodexLoginProfile(claim) ? [] : ['--model', codexModel(claim)];
|
||||||
|
|
||||||
const writeJsonFile = async (filePath: string, content: string) => {
|
const writeJsonFile = async (filePath: string, content: string) => {
|
||||||
let normalized = content.trim();
|
let normalized = content.trim();
|
||||||
@@ -696,20 +697,26 @@ const runCodexTurn = async (args: {
|
|||||||
codexSessionId: workspace.codexSessionId,
|
codexSessionId: workspace.codexSessionId,
|
||||||
});
|
});
|
||||||
const command = workspace.codexSessionId
|
const command = workspace.codexSessionId
|
||||||
? commandToShell(
|
? [
|
||||||
`codex exec resume --json${codexModelFlag(
|
'codex',
|
||||||
workspace.claim,
|
'exec',
|
||||||
)} --dangerously-bypass-approvals-and-sandbox ${quoteShell(
|
'resume',
|
||||||
workspace.codexSessionId,
|
'--json',
|
||||||
)} ${quoteShell(prompt)}`,
|
...codexModelArgs(workspace.claim),
|
||||||
)
|
'--dangerously-bypass-approvals-and-sandbox',
|
||||||
: commandToShell(
|
workspace.codexSessionId,
|
||||||
`codex exec --json${codexModelFlag(
|
prompt,
|
||||||
workspace.claim,
|
]
|
||||||
)} --dangerously-bypass-approvals-and-sandbox --cd ${quoteShell(
|
: [
|
||||||
codexContainerRepo,
|
'codex',
|
||||||
)} ${quoteShell(prompt)}`,
|
'exec',
|
||||||
);
|
'--json',
|
||||||
|
...codexModelArgs(workspace.claim),
|
||||||
|
'--dangerously-bypass-approvals-and-sandbox',
|
||||||
|
'--cd',
|
||||||
|
codexContainerRepo,
|
||||||
|
prompt,
|
||||||
|
];
|
||||||
const aiEnv = providerEnvironment(workspace.claim, codexContainerWorkspace);
|
const aiEnv = providerEnvironment(workspace.claim, codexContainerWorkspace);
|
||||||
const secretEnv = Object.fromEntries(
|
const secretEnv = Object.fromEntries(
|
||||||
workspace.claim.secrets.map((secret) => [secret.name, secret.value]),
|
workspace.claim.secrets.map((secret) => [secret.name, secret.value]),
|
||||||
@@ -734,12 +741,17 @@ const runCodexTurn = async (args: {
|
|||||||
}
|
}
|
||||||
},
|
},
|
||||||
onStderrLine: async (line) => {
|
onStderrLine: async (line) => {
|
||||||
if (line.trim()) {
|
const trimmed = line.trim();
|
||||||
|
if (
|
||||||
|
trimmed &&
|
||||||
|
trimmed !== 'Reading additional input from stdin...' &&
|
||||||
|
!trimmed.includes('`[features].codex_hooks` is deprecated')
|
||||||
|
) {
|
||||||
await appendEvent(
|
await appendEvent(
|
||||||
workspace.claim.job._id,
|
workspace.claim.job._id,
|
||||||
'info',
|
'info',
|
||||||
'plan',
|
'plan',
|
||||||
truncate(line, 10_000),
|
truncate(trimmed, 10_000),
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
@@ -1034,6 +1046,9 @@ const recordChangedFiles = async (
|
|||||||
workspace.repoDir,
|
workspace.repoDir,
|
||||||
workspace.redact,
|
workspace.redact,
|
||||||
);
|
);
|
||||||
|
const signature = JSON.stringify({ diff, changes });
|
||||||
|
if (signature === workspace.lastRecordedDiffSignature) return;
|
||||||
|
workspace.lastRecordedDiffSignature = signature;
|
||||||
for (const change of changes) {
|
for (const change of changes) {
|
||||||
await recordWorkspaceChange({
|
await recordWorkspaceChange({
|
||||||
jobId: workspace.claim.job._id,
|
jobId: workspace.claim.job._id,
|
||||||
@@ -1235,7 +1250,9 @@ const runClaim = async (claim: Claim) => {
|
|||||||
});
|
});
|
||||||
await appendEvent(jobId, 'info', 'plan', 'Interactive workspace is ready.');
|
await appendEvent(jobId, 'info', 'plan', 'Interactive workspace is ready.');
|
||||||
|
|
||||||
await sendWorkspaceMessage(jobId, systemPromptForJob(claim));
|
await sendWorkspaceMessage(jobId, systemPromptForJob(claim), {
|
||||||
|
recordUserMessage: false,
|
||||||
|
});
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
const message = error instanceof Error ? error.message : String(error);
|
const message = error instanceof Error ? error.message : String(error);
|
||||||
await appendEvent(
|
await appendEvent(
|
||||||
@@ -1428,18 +1445,24 @@ export const replyToInteraction = async (
|
|||||||
return { success: true };
|
return { success: true };
|
||||||
};
|
};
|
||||||
|
|
||||||
export const sendWorkspaceMessage = async (jobId: string, prompt: string) => {
|
export const sendWorkspaceMessage = async (
|
||||||
|
jobId: string,
|
||||||
|
prompt: string,
|
||||||
|
options: { recordUserMessage?: boolean } = {},
|
||||||
|
) => {
|
||||||
const workspace = resolveWorkspace(jobId);
|
const workspace = resolveWorkspace(jobId);
|
||||||
const { claim, redact } = workspace;
|
const { claim, redact } = workspace;
|
||||||
if (workspace.agentTurnActive) {
|
if (workspace.agentTurnActive) {
|
||||||
throw new Error('Wait for the current agent turn to finish or abort it.');
|
throw new Error('Wait for the current agent turn to finish or abort it.');
|
||||||
}
|
}
|
||||||
await appendMessage({
|
if (options.recordUserMessage ?? true) {
|
||||||
jobId: claim.job._id,
|
await appendMessage({
|
||||||
role: 'user',
|
jobId: claim.job._id,
|
||||||
status: 'completed',
|
role: 'user',
|
||||||
content: prompt,
|
status: 'completed',
|
||||||
});
|
content: prompt,
|
||||||
|
});
|
||||||
|
}
|
||||||
await appendEvent(claim.job._id, 'info', 'plan', 'Sending message to agent.');
|
await appendEvent(claim.job._id, 'info', 'plan', 'Sending message to agent.');
|
||||||
|
|
||||||
let assistantMessageId: Id<'agentJobMessages'> | undefined;
|
let assistantMessageId: Id<'agentJobMessages'> | undefined;
|
||||||
|
|||||||
@@ -95,7 +95,8 @@ describe('agent event normalization', () => {
|
|||||||
),
|
),
|
||||||
).toContainEqual({
|
).toContainEqual({
|
||||||
kind: 'assistant_delta',
|
kind: 'assistant_delta',
|
||||||
content: 'I updated the auth provider.',
|
content: 'I updated the auth provider.\n\n',
|
||||||
|
externalMessageId: 'item-1',
|
||||||
});
|
});
|
||||||
|
|
||||||
expect(
|
expect(
|
||||||
@@ -125,6 +126,24 @@ describe('agent event normalization', () => {
|
|||||||
kind: 'error',
|
kind: 'error',
|
||||||
message: '{\n "message": "request failed"\n}',
|
message: '{\n "message": "request failed"\n}',
|
||||||
});
|
});
|
||||||
|
|
||||||
|
expect(
|
||||||
|
normalizeCodexJsonLine(
|
||||||
|
JSON.stringify({
|
||||||
|
type: 'item.completed',
|
||||||
|
item: {
|
||||||
|
id: 'item-warning',
|
||||||
|
type: 'error',
|
||||||
|
message:
|
||||||
|
'`[features].codex_hooks` is deprecated. Use `[features].hooks` instead.',
|
||||||
|
},
|
||||||
|
}),
|
||||||
|
),
|
||||||
|
).toContainEqual({
|
||||||
|
kind: 'status',
|
||||||
|
status:
|
||||||
|
'`[features].codex_hooks` is deprecated. Use `[features].hooks` instead.',
|
||||||
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
test('normalizes Codex tool item lifecycle events', () => {
|
test('normalizes Codex tool item lifecycle events', () => {
|
||||||
@@ -141,7 +160,7 @@ describe('agent event normalization', () => {
|
|||||||
),
|
),
|
||||||
).toContainEqual({
|
).toContainEqual({
|
||||||
kind: 'tool_started',
|
kind: 'tool_started',
|
||||||
name: 'local_shell_call',
|
name: 'Command',
|
||||||
input: 'bash -lc rg Authentik',
|
input: 'bash -lc rg Authentik',
|
||||||
externalMessageId: 'tool-1',
|
externalMessageId: 'tool-1',
|
||||||
});
|
});
|
||||||
@@ -160,10 +179,30 @@ describe('agent event normalization', () => {
|
|||||||
),
|
),
|
||||||
).toContainEqual({
|
).toContainEqual({
|
||||||
kind: 'tool_completed',
|
kind: 'tool_completed',
|
||||||
name: 'local_shell_call',
|
name: 'Command',
|
||||||
output: 'apps/web/auth.ts',
|
output: 'apps/web/auth.ts',
|
||||||
externalMessageId: 'tool-1',
|
externalMessageId: 'tool-1',
|
||||||
});
|
});
|
||||||
|
|
||||||
|
expect(
|
||||||
|
normalizeCodexJsonLine(
|
||||||
|
JSON.stringify({
|
||||||
|
type: 'item.completed',
|
||||||
|
item: {
|
||||||
|
id: 'tool-2',
|
||||||
|
type: 'exec_command',
|
||||||
|
command: 'cat package.json',
|
||||||
|
aggregated_output: '{"scripts":{"build":"turbo build"}}',
|
||||||
|
exit_code: 0,
|
||||||
|
},
|
||||||
|
}),
|
||||||
|
),
|
||||||
|
).toContainEqual({
|
||||||
|
kind: 'tool_completed',
|
||||||
|
name: 'Command',
|
||||||
|
output: '{"scripts":{"build":"turbo build"}}',
|
||||||
|
externalMessageId: 'tool-2',
|
||||||
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
test('normalizes OpenCode assistant, tool, and permission events', () => {
|
test('normalizes OpenCode assistant, tool, and permission events', () => {
|
||||||
|
|||||||
Reference in New Issue
Block a user