Menu
popagent
publicLatest change 2fb6f198c4c71ef37dffc8ac5dca8482068a7bfc - Add governed AkurAI Build maintenance by AkurAI Build
import { createStep, createWorkflow } from "@mastra/core/workflows";
import type { Agent } from "@mastra/core/agent";
import { z } from "zod";
import {
agent as orchistrator,
implementer,
researcher,
reviewer,
} from "./agent";
import { AGENT_CONTEXT_KEYS, agentContextValue } from "./agent-context";
import type { AgentExecutionRuntime } from "./agent-execution";
import {
AUTONOMY_WORKFLOW_PHASES,
type AgentTask,
type AgentWorkflowDetail,
type AgentWorkflowPhaseId,
} from "./api-types";
import { SelfUpdatePolicyError, type SelfUpdateRun } from "./self-update-policy";
const MAX_EVIDENCE = 12;
const MAX_EVIDENCE_LENGTH = 500;
const MAX_HANDOFF_TEXT = 4_000;
const phaseLabels: Record<AgentWorkflowPhaseId, string> = {
prepare: "Prepare",
research: "Research",
decide: "Decide",
implement: "Implement",
"inspect-change": "Inspect change",
verify: "Verify",
review: "Review",
commit: "Commit",
};
const snapshotSchema = z.object({
head: z.string().min(1).max(128),
branch: z.string().min(1).max(256),
remotes: z.string().min(1).max(128),
});
const runSchema = z.object({
taskId: z.string().min(1).max(128),
workspaceId: z.string().min(1).max(256),
repositoryPath: z.string().min(1).max(2_048),
path: z.string().min(1).max(2_048),
branch: z.string().min(1).max(256),
snapshot: snapshotSchema,
cloneRemotes: z.string().min(1).max(128),
delegations: z.array(z.string().max(64)).max(8),
deploymentEnabled: z.boolean(),
source: z.literal("self-update"),
});
type SerializedRun = z.infer<typeof runSchema>;
const researchSchema = z.object({
summary: z.string().trim().min(1).max(MAX_HANDOFF_TEXT),
evidence: z.array(z.string().trim().min(1).max(MAX_EVIDENCE_LENGTH)).max(MAX_EVIDENCE),
relevantPaths: z.array(z.string().trim().min(1).max(256)).max(64),
noOp: z.boolean(),
});
const decisionSchema = z.object({
rationale: z.string().trim().min(1).max(MAX_HANDOFF_TEXT),
noOp: z.boolean(),
verificationCommand: z.string().trim().max(256).nullable(),
});
const implementationSchema = z.object({
summary: z.string().trim().max(MAX_HANDOFF_TEXT),
changedPaths: z.array(z.string().trim().min(1).max(256)).max(64),
});
const inspectionSchema = z.object({
changedPaths: z.array(z.string().trim().min(1).max(256)).max(64),
diffCheck: z.string().max(MAX_HANDOFF_TEXT),
clean: z.boolean(),
});
const verificationSchema = z.object({
command: z.string().trim().min(1).max(256),
output: z.string().max(MAX_HANDOFF_TEXT),
passed: z.boolean(),
});
const reviewSchema = z.object({
summary: z.string().trim().min(1).max(MAX_HANDOFF_TEXT),
issues: z.array(z.string().trim().min(1).max(MAX_EVIDENCE_LENGTH)).max(MAX_EVIDENCE),
accepted: z.boolean(),
});
const inputSchema = z.object({
taskId: z.string().min(1).max(128),
workspaceId: z.string().min(1).max(256),
prompt: z.string().trim().min(1).max(32_000),
model: z.string().min(1).max(256),
sessionId: z.string().min(1).max(256),
turnId: z.string().min(1).max(128),
traceId: z.string().min(1).max(128),
});
type WorkflowInput = z.infer<typeof inputSchema>;
const stateSchema = z.object({
runId: z.string().min(1).max(128),
phase: z.enum(AUTONOMY_WORKFLOW_PHASES),
});
const prepareOutputSchema = z.object({ run: runSchema });
const researchOutputSchema = z.object({ run: runSchema, research: researchSchema });
const decideOutputSchema = z.object({ run: runSchema, research: researchSchema, decision: decisionSchema });
const implementOutputSchema = z.object({
run: runSchema,
research: researchSchema,
decision: decisionSchema,
implementation: implementationSchema,
});
const inspectOutputSchema = z.object({
run: runSchema,
research: researchSchema,
decision: decisionSchema,
implementation: implementationSchema,
inspection: inspectionSchema,
});
const verifyOutputSchema = z.object({
run: runSchema,
research: researchSchema,
decision: decisionSchema,
implementation: implementationSchema,
inspection: inspectionSchema,
verification: verificationSchema,
});
const reviewOutputSchema = z.object({
run: runSchema,
research: researchSchema,
decision: decisionSchema,
implementation: implementationSchema,
inspection: inspectionSchema,
verification: verificationSchema,
review: reviewSchema,
});
const outputSchema = z.object({ output: z.string() });
type WorkflowReporter = (detail: AgentWorkflowDetail) => Promise<void>;
export type SelfUpdateWorkflowAgents = {
orchistrator: Agent;
researcher: Agent;
implementer: Agent;
reviewer: Agent;
};
export type SelfUpdateWorkflowDependencies = {
execution: AgentExecutionRuntime;
policy: {
begin(task: AgentTask): Promise<SelfUpdateRun>;
observeDelegation(run: SelfUpdateRun, agentId: string, success: boolean): void;
inspect(run: SelfUpdateRun): Promise<{ changedPaths: string[]; diffCheck: string; clean: boolean }>;
verify(run: SelfUpdateRun, command: string): Promise<string>;
commit(run: SelfUpdateRun): Promise<string>;
accept(run: SelfUpdateRun, signal?: AbortSignal): Promise<void>;
reject(run: SelfUpdateRun): Promise<void>;
};
agents?: Partial<SelfUpdateWorkflowAgents>;
report: WorkflowReporter;
signal: AbortSignal;
};
function toSerializedRun(run: SelfUpdateRun): SerializedRun {
return {
taskId: run.taskId,
workspaceId: run.workspaceId,
repositoryPath: run.repositoryPath,
path: run.path,
branch: run.branch,
snapshot: run.snapshot,
cloneRemotes: run.cloneRemotes,
delegations: [...run.delegations],
deploymentEnabled: run.deploymentEnabled,
source: "self-update",
};
}
function toRuntimeRun(run: SerializedRun): SelfUpdateRun {
return { ...run, delegations: new Set(run.delegations) };
}
function boundedEvidence(values: string[]): string[] {
return [...new Set(values.map((value) => value.trim()).filter(Boolean))]
.slice(0, MAX_EVIDENCE)
.map((value) => value.slice(0, MAX_EVIDENCE_LENGTH));
}
function initialDetail(runId: string, correlationId: string): AgentWorkflowDetail {
return {
runId,
correlationId,
state: "running",
currentPhase: null,
phases: AUTONOMY_WORKFLOW_PHASES.map((id) => ({
id,
label: phaseLabels[id],
state: "waiting" as const,
evidence: [],
startedAt: null,
completedAt: null,
error: null,
})),
};
}
function phaseError(phase: AgentWorkflowPhaseId, error: unknown): Error {
const message = error instanceof Error ? error.message : String(error);
const name = `SelfUpdate${phase.replace(/(^|-)([a-z])/g, (_, __, letter: string) => letter.toUpperCase())}Error`;
const wrapped = new Error(`${phaseLabels[phase]} phase failed: ${message}`, { cause: error });
wrapped.name = name;
return wrapped;
}
export async function runSelfUpdateWorkflow(
task: AgentTask,
turnId: string,
traceId: string,
dependencies: SelfUpdateWorkflowDependencies,
): Promise<string> {
const detail = initialDetail(traceId, traceId);
let activeRun: SerializedRun | undefined;
await dependencies.report(detail);
const input: WorkflowInput = {
taskId: task.id,
workspaceId: task.workspaceId,
prompt: task.prompt,
model: task.model,
sessionId: task.sessionId ?? `task:${task.id}`,
turnId,
traceId,
};
const workflowAgents: SelfUpdateWorkflowAgents = {
orchistrator: dependencies.agents?.orchistrator ?? orchistrator,
researcher: dependencies.agents?.researcher ?? researcher,
implementer: dependencies.agents?.implementer ?? implementer,
reviewer: dependencies.agents?.reviewer ?? reviewer,
};
const checkCancelled = () => dependencies.signal.throwIfAborted();
const update = async (
phase: AgentWorkflowPhaseId,
state: "active" | "complete" | "failed" | "cancelled" | "skipped",
evidence: string[] = [],
error: string | null = null,
) => {
const current = detail.phases.find((item) => item.id === phase)!;
const now = new Date().toISOString();
current.state = state;
current.evidence = boundedEvidence(evidence);
current.error = error?.slice(0, MAX_EVIDENCE_LENGTH) ?? null;
if (state === "active") current.startedAt = now;
if (["complete", "failed", "cancelled", "skipped"].includes(state)) current.completedAt = now;
detail.currentPhase = state === "active" ? phase : detail.currentPhase;
await dependencies.report({ ...detail, phases: detail.phases.map((item) => ({ ...item, evidence: [...item.evidence] })) });
};
const skipLater = async (from: AgentWorkflowPhaseId) => {
const index = AUTONOMY_WORKFLOW_PHASES.indexOf(from);
for (const id of AUTONOMY_WORKFLOW_PHASES.slice(index + 1)) {
if (detail.phases.find((item) => item.id === id)?.state === "waiting") await update(id, "skipped", ["No-op branch"]);
}
};
const runAgent = async <T>(
phase: AgentWorkflowPhaseId,
target: Agent,
prompt: string,
schema: z.ZodType<T>,
specialist: boolean,
workspacePath: string,
): Promise<T> => {
checkCancelled();
const prepared = await dependencies.execution.prepare({
sessionId: input.sessionId,
turnId,
traceId,
model: input.model,
workspaceId: input.workspaceId,
executionSource: "self-update",
taskId: task.id,
selfUpdateWorkspacePath: workspacePath,
prompt,
skipManagedHooks: true,
});
const runtimeSettings = agentContextValue(prepared.options.requestContext, AGENT_CONTEXT_KEYS.runtimeSettings);
// Contained runs intentionally do not carry chat memory mutation or browser
// state. Forking keeps the production agent graph isolated while disabling
// the browser state-signal processor, which requires a memory-backed thread.
const delegated = target.__fork();
delegated.setBrowser(undefined);
const result = await delegated.generate(prompt, {
...prepared.options,
maxSteps: specialist ? runtimeSettings?.specialistMaxSteps : runtimeSettings?.supervisorMaxSteps,
abortSignal: dependencies.signal,
structuredOutput: { schema },
runId: `${traceId}-${phase}`,
});
return result.object as T;
};
const prepare = createStep({
id: "prepare",
inputSchema,
outputSchema: prepareOutputSchema,
stateSchema,
execute: async ({ setState }) => {
await update("prepare", "active");
try {
checkCancelled();
const run = await dependencies.policy.begin(task);
await setState({ runId: traceId, phase: "prepare" });
await update("prepare", "complete", [
`HEAD ${run.snapshot.head}`,
`Branch ${run.snapshot.branch}`,
`Remote digest ${run.snapshot.remotes}`,
]);
const serialized = toSerializedRun(run);
activeRun = serialized;
return { run: serialized };
} catch (error) {
await update("prepare", dependencies.signal.aborted ? "cancelled" : "failed", [], String(error));
throw phaseError("prepare", error);
}
},
});
const researchStep = createStep({
id: "research",
inputSchema: prepareOutputSchema,
outputSchema: researchOutputSchema,
stateSchema,
execute: async ({ inputData, setState }) => {
await update("research", "active");
try {
const evidence = await dependencies.policy.inspect(toRuntimeRun(inputData.run));
const research = await runAgent("research", workflowAgents.researcher,
`Research the requested contained self-update. Do not edit files or commit. Inspect only the repository and return strict JSON matching the schema. The work request is:\n${input.prompt}\nDeterministic checkout evidence:\n${JSON.stringify(evidence)}`,
researchSchema,
true,
inputData.run.path);
await setState({ runId: traceId, phase: "research" });
await update("research", "complete", [research.summary, ...research.evidence]);
return { run: inputData.run, research };
} catch (error) {
await update("research", dependencies.signal.aborted ? "cancelled" : "failed", [], String(error));
throw phaseError("research", error);
}
},
});
const decideStep = createStep({
id: "decide",
inputSchema: researchOutputSchema,
outputSchema: decideOutputSchema,
stateSchema,
execute: async ({ inputData, setState }) => {
await update("decide", "active");
try {
const decision = await runAgent("decide", workflowAgents.orchistrator,
`Decide the contained self-update from this typed Researcher handoff. Do not edit files. Return strict JSON. A no-op is allowed only when the request requires no source change. Choose one repository-owned verification command from the allowed local commands, or null for a no-op.\nResearch handoff:\n${JSON.stringify(inputData.research)}`,
decisionSchema,
false,
inputData.run.path);
const normalized = {
...decision,
noOp: decision.noOp || inputData.research.noOp,
verificationCommand: decision.noOp ? null : decision.verificationCommand ?? "git diff --check",
};
if (normalized.noOp) await skipLater("decide");
await setState({ runId: traceId, phase: "decide" });
await update("decide", "complete", [normalized.rationale, normalized.noOp ? "No-op branch" : "Change branch"]);
return { ...inputData, decision: normalized };
} catch (error) {
await update("decide", dependencies.signal.aborted ? "cancelled" : "failed", [], String(error));
throw phaseError("decide", error);
}
},
});
const implementStep = createStep({
id: "implement",
inputSchema: decideOutputSchema,
outputSchema: implementOutputSchema,
stateSchema,
execute: async ({ inputData, setState }) => {
if (inputData.decision.noOp) {
return {
...inputData,
implementation: { summary: "Skipped for no-op", changedPaths: [] },
};
}
await update("implement", "active");
try {
const implementation = await runAgent("implement", workflowAgents.implementer,
`Implement the requested change in the contained workspace. Follow the Researcher evidence and Decision handoff exactly. Edit only allowed source paths. Do not commit, publish, deploy, or run unrelated commands. Return strict JSON after editing.\nRequest: ${input.prompt}\nResearch: ${JSON.stringify(inputData.research)}\nDecision: ${JSON.stringify(inputData.decision)}`,
implementationSchema,
true,
inputData.run.path);
const run = toRuntimeRun(inputData.run);
dependencies.policy.observeDelegation(run, "implementer", true);
const nextRun = toSerializedRun(run);
await setState({ runId: traceId, phase: "implement" });
await update("implement", "complete", [implementation.summary, ...implementation.changedPaths]);
return { ...inputData, run: nextRun, implementation };
} catch (error) {
await update("implement", dependencies.signal.aborted ? "cancelled" : "failed", [], String(error));
throw phaseError("implement", error);
}
},
});
const inspectStep = createStep({
id: "inspect-change",
inputSchema: implementOutputSchema,
outputSchema: inspectOutputSchema,
stateSchema,
execute: async ({ inputData, setState }) => {
if (inputData.decision.noOp) {
return { ...inputData, inspection: { changedPaths: [], diffCheck: "No-op", clean: true } };
}
await update("inspect-change", "active");
try {
checkCancelled();
const inspection = await dependencies.policy.inspect(toRuntimeRun(inputData.run));
if (inspection.clean) throw new SelfUpdatePolicyError("Implementation produced no source change");
await setState({ runId: traceId, phase: "inspect-change" });
await update("inspect-change", "complete", [...inspection.changedPaths, inspection.diffCheck]);
return { ...inputData, inspection };
} catch (error) {
await update("inspect-change", dependencies.signal.aborted ? "cancelled" : "failed", [], String(error));
throw phaseError("inspect-change", error);
}
},
});
const verifyStep = createStep({
id: "verify",
inputSchema: inspectOutputSchema,
outputSchema: verifyOutputSchema,
stateSchema,
execute: async ({ inputData, setState }) => {
if (inputData.decision.noOp) {
return { ...inputData, verification: { command: "none", output: "No-op", passed: true } };
}
await update("verify", "active");
try {
checkCancelled();
const command = inputData.decision.verificationCommand ?? "git diff --check";
const output = await dependencies.policy.verify(toRuntimeRun(inputData.run), command);
await setState({ runId: traceId, phase: "verify" });
await update("verify", "complete", [command, output]);
return { ...inputData, verification: { command, output, passed: true } };
} catch (error) {
await update("verify", dependencies.signal.aborted ? "cancelled" : "failed", [], String(error));
throw phaseError("verify", error);
}
},
});
const reviewStep = createStep({
id: "review",
inputSchema: verifyOutputSchema,
outputSchema: reviewOutputSchema,
stateSchema,
execute: async ({ inputData, setState }) => {
if (inputData.decision.noOp) {
return { ...inputData, review: { summary: "Skipped for no-op", issues: [], accepted: true } };
}
await update("review", "active");
try {
const review = await runAgent("review", workflowAgents.reviewer,
`Review the contained self-update read-only. Do not edit or commit. Reject if verification evidence is missing or the change exceeds the request. Return strict JSON.\nRequest: ${input.prompt}\nResearch: ${JSON.stringify(inputData.research)}\nImplementation: ${JSON.stringify(inputData.implementation)}\nInspection: ${JSON.stringify(inputData.inspection)}\nVerification: ${JSON.stringify(inputData.verification)}`,
reviewSchema,
true,
inputData.run.path);
const run = toRuntimeRun(inputData.run);
dependencies.policy.observeDelegation(run, "reviewer", review.accepted);
const nextRun = toSerializedRun(run);
if (!review.accepted) throw new SelfUpdatePolicyError(`Reviewer rejected the change: ${review.summary}`);
await setState({ runId: traceId, phase: "review" });
await update("review", "complete", [review.summary, ...review.issues]);
return { ...inputData, run: nextRun, review };
} catch (error) {
await update("review", dependencies.signal.aborted ? "cancelled" : "failed", [], String(error));
throw phaseError("review", error);
}
},
});
const commitStep = createStep({
id: "commit",
inputSchema: reviewOutputSchema,
outputSchema,
stateSchema,
execute: async ({ inputData }) => {
if (inputData.decision.noOp) {
await dependencies.policy.accept(toRuntimeRun(inputData.run), dependencies.signal);
activeRun = undefined;
return { output: "Self-update completed with no source changes" };
}
await update("commit", "active");
try {
checkCancelled();
const commit = await dependencies.policy.commit(toRuntimeRun(inputData.run));
checkCancelled();
await dependencies.policy.accept(toRuntimeRun(inputData.run), dependencies.signal);
activeRun = undefined;
await update("commit", "complete", [commit]);
return { output: `Self-update accepted local commit ${commit}` };
} catch (error) {
await update("commit", dependencies.signal.aborted ? "cancelled" : "failed", [], String(error));
throw phaseError("commit", error);
}
},
});
const workflow = createWorkflow({
id: `self-update-${task.id}`,
inputSchema,
outputSchema,
stateSchema,
})
.then(prepare)
.then(researchStep)
.then(decideStep)
.then(implementStep)
.then(inspectStep)
.then(verifyStep)
.then(reviewStep)
.then(commitStep)
.commit();
const run = await workflow.createRun({ runId: traceId, resourceId: task.id });
try {
const result = await run.start({ inputData: input, initialState: { runId: traceId, phase: "prepare" } });
if (result.status !== "success") {
const workflowError = "error" in result ? result.error : undefined;
throw workflowError instanceof Error ? workflowError : new Error(`Self-update workflow ${result.status}`);
}
detail.state = "completed";
detail.currentPhase = null;
await dependencies.report(detail);
return result.result.output;
} catch (error) {
detail.state = dependencies.signal.aborted ? "cancelled" : "failed";
detail.currentPhase = null;
if (activeRun) {
const runToRelease = activeRun;
activeRun = undefined;
await dependencies.policy.reject(toRuntimeRun(runToRelease)).catch(() => undefined);
}
if (dependencies.signal.aborted) {
for (const phase of detail.phases) {
if (phase.state === "active" || phase.state === "waiting") {
phase.state = phase.state === "active" ? "cancelled" : "skipped";
phase.completedAt = new Date().toISOString();
}
}
}
await dependencies.report(detail).catch(() => undefined);
throw error;
}
}