Menu
popagent
publicLatest change 9f18b8282153586896c8d904d2999133cb9fdde4 - Stream autonomous workflow activity by AkurAI Build
import { createStep, createWorkflow } from "@mastra/core/workflows";
import type { Agent } from "@mastra/core/agent";
import type { IterationCompleteContext } from "@mastra/core/agent";
import { RequestContext } from "@mastra/core/request-context";
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,
SELF_UPDATE_REVIEW_DIFF_MAX_BYTES,
type SelfUpdateInspection,
type SelfUpdateRun,
} from "./self-update-policy";
const MAX_EVIDENCE = 12;
const MAX_EVIDENCE_LENGTH = 500;
const MAX_HANDOFF_TEXT = 4_000;
const EMPTY_DIFF_DIGEST = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855";
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),
reviewedDiffDigest: z.string().length(64).nullable(),
deploymentEnabled: z.boolean(),
source: z.literal("self-update"),
});
type SerializedRun = z.infer<typeof runSchema>;
function evidenceItems(value: unknown): unknown {
if (Array.isArray(value)) return value;
if (!value || typeof value !== "object") return value;
return Object.entries(value).map(([key, item]) =>
`${key}: ${typeof item === "string" ? item : JSON.stringify(item)}`);
}
function researchInput(value: unknown): unknown {
if (!value || typeof value !== "object" || Array.isArray(value)) return value;
if ("summary" in value) return value;
const nested = Object.values(value).find((item) =>
item && typeof item === "object" && !Array.isArray(item) && "summary" in item);
return nested ?? value;
}
const researchSchema = z.preprocess(researchInput, z.object({
summary: z.string().trim().min(1).max(MAX_HANDOFF_TEXT),
evidence: z.preprocess(evidenceItems,
z.array(z.string().trim().min(1).max(MAX_EVIDENCE_LENGTH)).max(64))
.transform((values) => [...new Set(values)].slice(0, 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),
diffEvidence: z.string().max(SELF_UPDATE_REVIEW_DIFF_MAX_BYTES),
diffDigest: z.string().length(64),
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(64)
.transform((values) => [...new Set(values)].slice(0, MAX_EVIDENCE)),
accepted: z.boolean(),
diffDigest: z.string().length(64),
});
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, diffDigest?: string): void;
inspect(run: SelfUpdateRun): Promise<SelfUpdateInspection>;
verify(run: SelfUpdateRun, command: string, signal?: AbortSignal): Promise<string>;
commit(run: SelfUpdateRun, expectedDiffDigest: string): Promise<string>;
accept(run: SelfUpdateRun, signal?: AbortSignal): Promise<void>;
reject(run: SelfUpdateRun): Promise<void>;
};
agents?: Partial<SelfUpdateWorkflowAgents>;
report: WorkflowReporter;
observeIteration?: (iteration: IterationCompleteContext) => Promise<void>;
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],
reviewedDiffDigest: run.reviewedDiffDigest ?? null,
deploymentEnabled: run.deploymentEnabled,
source: "self-update",
};
}
function toRuntimeRun(run: SerializedRun): SelfUpdateRun {
const { reviewedDiffDigest, ...rest } = run;
return {
...rest,
...(reviewedDiffDigest ? { reviewedDiffDigest } : {}),
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 parseStructuredText(text: string): unknown {
const unfenced = text.trim().replace(/^```(?:json)?\s*/i, "").replace(/\s*```$/, "");
const start = unfenced.indexOf("{");
if (start < 0) return JSON.parse(unfenced);
let depth = 0;
let quoted = false;
let escaped = false;
for (let index = start; index < unfenced.length; index += 1) {
const character = unfenced[index]!;
if (quoted) {
if (escaped) escaped = false;
else if (character === "\\") escaped = true;
else if (character === "\"") quoted = false;
continue;
}
if (character === "\"") quoted = true;
else if (character === "{") depth += 1;
else if (character === "}" && --depth === 0) return JSON.parse(unfenced.slice(start, index + 1));
}
return JSON.parse(unfenced.slice(start));
}
function retryablePhaseError(error: unknown): boolean {
const message = error instanceof Error ? error.message : String(error);
return /socket|ECONNRESET|ECONNREFUSED|fetch failed|timed? ?out|Unexpected EOF|Unexpected identifier/i.test(message);
}
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 function isContainmentDeniedEvidence(text: string): boolean {
return /\bcontainment\b.{0,80}\bden(?:y|ied|ial)\b|\bden(?:y|ied|ial)\b.{0,80}\bcontainment\b/i.test(text);
}
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,
onIterationComplete: dependencies.observeIteration,
});
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 structuringContext = new RequestContext();
const structuringModel = await (dependencies.agents?.orchistrator ?? orchistrator).getModel({
requestContext: structuringContext,
});
let lastError: unknown;
for (let attempt = 0; attempt < 2; attempt += 1) {
try {
const result = await delegated.stream(prompt, {
...prepared.options,
maxSteps: specialist ? runtimeSettings?.specialistMaxSteps : runtimeSettings?.supervisorMaxSteps,
abortSignal: dependencies.signal,
structuredOutput: { schema, jsonPromptInjection: "auto", model: structuringModel },
runId: `${traceId}-${phase}`,
});
const object = await result.object;
if (object !== undefined) return schema.parse(object);
return schema.parse(parseStructuredText(await result.text));
} catch (error) {
lastError = error;
if (attempt || !retryablePhaseError(error)) throw error;
}
}
throw lastError;
};
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);
const researchText = [research.summary, ...research.evidence].join("\n");
if (isContainmentDeniedEvidence(researchText)) {
throw new SelfUpdatePolicyError("Repository inspection was blocked by self-update containment");
}
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 noOp = decision.noOp && inputData.research.noOp;
const normalized = {
...decision,
noOp,
verificationCommand: 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",
diffEvidence: "",
diffDigest: EMPTY_DIFF_DIGEST,
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, dependencies.signal);
await setState({ runId: traceId, phase: "verify" });
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,
diffDigest: EMPTY_DIFF_DIGEST,
},
};
}
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. The inspection includes the complete bounded patch, including deletions. Approval MUST echo the exact inspection diffDigest in the response; reject if it is absent or does not match. 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);
if (review.diffDigest !== inputData.inspection.diffDigest) {
throw new SelfUpdatePolicyError("Reviewer approval digest does not match the inspected diff");
}
const run = toRuntimeRun(inputData.run);
run.reviewedDiffDigest = inputData.inspection.diffDigest;
dependencies.policy.observeDelegation(run, "reviewer", review.accepted, review.diffDigest);
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 verified no-op with no source changes" };
}
await update("commit", "active");
try {
checkCancelled();
const commit = await dependencies.policy.commit(toRuntimeRun(inputData.run), inputData.inspection.diffDigest);
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: unknown = "error" in result ? result.error : undefined;
if (workflowError instanceof Error) throw workflowError;
if (workflowError && typeof workflowError === "object" && "message" in workflowError
&& typeof workflowError.message === "string") {
const error = new Error(workflowError.message);
if ("name" in workflowError && typeof workflowError.name === "string") error.name = workflowError.name;
throw error;
}
throw new Error(typeof workflowError === "string" ? workflowError : `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;
}
}