puter/zopu-code
1
0
Fork 0
Code Issues 3 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

feat: consolidate agent execution stack

Browse Source
This commit is contained in:
-Puter
2026-07-31 18:26:41 +05:30
parent c644ec8d01
commit ce1f33aa69
68 changed files with 2196 additions and 3827 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/agents/package.json
+8 -11
View File
@@ -7,29 +7,26 @@
"build": "node --env-file=../../.env node_modules/@flue/cli/bin/flue.mjs build",
"start": "node --env-file=../../.env dist/server.mjs",
"check-types": "tsc --noEmit",
"dev": "node --env-file=../../.env node_modules/@flue/cli/bin/flue.mjs dev",
"dev:tailscale": "node --env-file=../../.env node_modules/@flue/cli/bin/flue.mjs dev",
"dev": "node --env-file=../../.env node_modules/@flue/cli/bin/flue.mjs dev --port 3585",
"dev:engine": "node --env-file=../../.env scripts/start-engine.mjs",
"run": "node --env-file=../../.env node_modules/@flue/cli/bin/flue.mjs run",
"runner": "bun --env-file=../../.env src/runner.ts",
"run:zopu": "node --env-file=../../.env node_modules/@flue/cli/bin/flue.mjs run zopu",
"run:work-planner": "node --env-file=../../.env node_modules/@flue/cli/bin/flue.mjs run work-planner"
"run:zopu": "node --env-file=../../.env node_modules/@flue/cli/bin/flue.mjs run zopu"
},
"dependencies": {
"@agentos-software/git": "0.3.3",
"@code/backend": "workspace:*",
"@code/env": "workspace:*",
"@code/primitives": "workspace:*",
"@flue/runtime": "latest",
"@rivet-dev/agentos": "0.2.14",
"@rivet-dev/agentos-core": "0.2.14",
"@flue/runtime": "npm:@rivet-dev/labs-flue-runtime@1.0.0-beta.9-rivet.2",
"@rivet-dev/agentos": "catalog:",
"@rivet-dev/agentos-core": "catalog:",
"convex": "catalog:",
"hono": "catalog:",
"rivetkit": "2.3.9",
"valibot": "catalog:"
},
"devDependencies": {
"@code/config": "workspace:*",
"@flue/cli": "latest",
"@flue/cli": "npm:@rivet-dev/labs-flue-cli@1.0.0-beta.9-rivet.2",
"@rivetkit/engine-cli": "2.3.9",
"@types/bun": "catalog:",
"@types/node": "catalog:",
"typescript": "catalog:"
packages/agents/scripts/start-engine.mjs
+13
View File
@@ -0,0 +1,13 @@
import { spawn } from "node:child_process";
import { getEnginePath } from "@rivetkit/engine-cli";
const enginePath = getEnginePath();
const proc = spawn(enginePath, ["start"], { stdio: "inherit" });
process.on("SIGINT", () => proc.kill("SIGINT"));
process.on("SIGTERM", () => proc.kill("SIGTERM"));
proc.on("exit", (code) => {
process.exit(code ?? 0);
});
packages/agents/src/admission-context.ts → packages/agents/src/adapters/admission-context.ts
View File
packages/agents/src/adapters/convex-client.ts
+18
View File
@@ -0,0 +1,18 @@
import { parseAgentEnv } from "@code/env/agent";
import { ConvexHttpClient } from "convex/browser";
export interface ConvexServiceClient {
readonly client: ConvexHttpClient;
readonly token: string;
}
/** Shared factory for service-authenticated ConvexHttpClient used by agent tools. */
export const createConvexServiceClient = (
runtimeEnv: Record<string, string | undefined>
): ConvexServiceClient => {
const env = parseAgentEnv(runtimeEnv);
return {
client: new ConvexHttpClient(env.CONVEX_URL),
token: env.FLUE_DB_TOKEN,
};
};
packages/agents/src/db.ts → packages/agents/src/adapters/flue-persistence.ts
+20 -51
View File
@@ -8,11 +8,9 @@
* build typed references against explicit DTO shapes, so the file typechecks
* against the installed runtime package without depending on codegen output.
*
* DTO conventions shared with the backend:
* - Submissions persist `inputJson` (the dispatch input or the direct input
* with image data stripped via `prepareDirectSubmission`) plus `chunksJson`
* for direct submissions (a `PersistedImageChunk[]`). Direct inputs are
* rehydrated with `hydratePersistedDirectSubmission`.
* - Submissions persist a shared input payload plus `chunksJson` for
* attachments stripped by `prepareSubmissionAttachments` and restored by
* `hydratePersistedSubmissionAttachments`.
* - Conversation batches persist a numeric `offset` and `recordsJson`.
* - Event entries persist a numeric `offset` and `dataJson`.
* - Attachments carry `attachmentJson` and raw `bytes` as `ArrayBuffer`.
@@ -36,13 +34,14 @@ import {
assertSupportedFlueSchemaVersion,
clampLimit,
copyAttachmentBytes,
createDispatchAgentSubmissionInput,
createSessionStorageKey,
decodeRunCursor,
encodeRunCursor,
formatOffset,
hydratePersistedDirectSubmission,
parseOffset,
prepareDirectSubmission,
hydratePersistedSubmissionAttachments,
prepareSubmissionAttachments,
verifyAttachmentBytes,
} from "@flue/runtime/adapter";
import type {
@@ -61,9 +60,8 @@ import type {
ConversationStreamMeta,
ConversationStreamReadResult,
ConversationStreamStore,
AgentSubmissionInput,
CreateRunInput,
DirectAgentSubmissionInput,
DispatchAgentSubmissionInput,
DispatchInput,
EndRunInput,
EventStreamMeta,
@@ -113,7 +111,7 @@ interface TokenArgs {
const TOKEN = (): string => env.FLUE_DB_TOKEN;
/** Trace carrier shape (W3C tracecontext). Not exported by Flue's adapter. */
type TraceCarrier = NonNullable<DirectAgentSubmissionInput["traceCarrier"]>;
type TraceCarrier = NonNullable<AgentSubmissionInput["traceCarrier"]>;
// ---------------------------------------------------------------------------
// Shared DTO shapes (wire types returned by the Convex backend)
@@ -651,9 +649,7 @@ const attachmentRefToWire = (ref: AttachmentRef): AttachmentRefWire => ({
/**
* Rehydrate a {@link SubmissionRow} into the {@link AgentSubmission} contract.
* `dispatch` inputs round-trip through `inputJson`; `direct` inputs are split
* into `inputJson` (images stripped) + `chunksJson` and rehydrated with Flue's
* `hydratePersistedDirectSubmission`.
* Submission attachments are stored separately from the shared persisted payload.
*/
const hydrateSubmission = (row: SubmissionRow): AgentSubmission => {
const baseSubmission: Omit<AgentSubmission, "input"> = {
@@ -682,29 +678,15 @@ const hydrateSubmission = (row: SubmissionRow): AgentSubmission => {
...(row.error === null ? {} : { error: row.error }),
...(row.ownerId === null ? {} : { ownerId: row.ownerId }),
};
const traceCarrier = safeJsonParse<TraceCarrier>(row.traceCarrierJson);
if (row.kind === "dispatch") {
const input = safeJsonParse<DispatchAgentSubmissionInput>(row.inputJson);
if (input === undefined) {
throw new Error(
`[flue] persisted dispatch submission ${row.submissionId} has empty inputJson`
);
}
const inputWithCarrier =
traceCarrier === undefined ? input : { ...input, traceCarrier };
return { ...baseSubmission, input: inputWithCarrier };
}
const stripped = safeJsonParse<DirectAgentSubmissionInput>(row.inputJson);
if (stripped === undefined) {
const persisted = safeJsonParse<AgentSubmissionInput>(row.inputJson);
if (persisted === undefined) {
throw new Error(
`[flue] persisted direct submission ${row.submissionId} has empty inputJson`
`[flue] persisted submission ${row.submissionId} has empty inputJson`
);
}
const chunks = safeJsonParse<PersistedChunkRow[]>(row.chunksJson) ?? [];
const hydrated = hydratePersistedDirectSubmission(stripped, chunks);
const hydrated = hydratePersistedSubmissionAttachments(persisted, chunks);
const traceCarrier = safeJsonParse<TraceCarrier>(row.traceCarrierJson);
const input =
traceCarrier === undefined ? hydrated : { ...hydrated, traceCarrier };
return { ...baseSubmission, input };
@@ -849,25 +831,14 @@ class ConvexAgentSubmissionStore implements AgentSubmissionStore {
}
async admitDispatch(input: DispatchInput): Promise<AgentDispatchAdmission> {
const submissionInput = createDispatchAgentSubmissionInput(input);
const sessionKey = createSessionStorageKey(input.id, "default", "default");
const submissionInput: Omit<
DispatchAgentSubmissionInput,
"kind" | "submissionId"
> = {
acceptedAt: input.acceptedAt,
agent: input.agent,
dispatchId: input.dispatchId,
id: input.id,
input: input.input,
};
const extracted = prepareSubmissionAttachments(submissionInput);
const res = await this.convex.mutation(admitSubmission, {
input: {
acceptedAt: parseAcceptedAt(input.acceptedAt, "admitDispatch"),
inputJson: jsonStringify({
kind: "dispatch",
submissionId: input.dispatchId,
...submissionInput,
}),
chunksJson: jsonStringify(extracted.chunks),
inputJson: jsonStringify(extracted.value),
kind: "dispatch",
sessionKey,
submissionId: input.dispatchId,
@@ -886,11 +857,9 @@ class ConvexAgentSubmissionStore implements AgentSubmissionStore {
return { kind: "conflict" };
}
async admitDirect(
input: DirectAgentSubmissionInput
): Promise<AgentSubmission> {
async admitDirect(input: AgentSubmissionInput): Promise<AgentSubmission> {
const sessionKey = createSessionStorageKey(input.id, "default", "default");
const extracted = prepareDirectSubmission(input);
const extracted = prepareSubmissionAttachments(input);
const admission = currentTurnAdmission();
const res = await this.convex.mutation(admitSubmission, {
input: {
packages/agents/src/agents/work-planner.ts
-30
View File
@@ -1,30 +0,0 @@
import { parseAgentEnv } from "@code/env/agent";
import { defineAgent } from "@flue/runtime";
import { createWorkPlannerTools } from "../tools/work-planner";
const INSTRUCTIONS = `You are Zopu's private Work Planner.
You may only submit typed proposals through tools:
- a Work Definition proposal;
- a Design Packet proposal containing 1-4 observable, independently verifiable slices;
- structured questions.
You must never approve a Definition or Design, change Work status, start execution, claim implementation, create Git changes, or claim that simulation implemented the Work. Convex validates and stores every proposal. Ask a focused structured question when a high-impact decision is unresolved.`;
export {
convexAgentRoute as attachments,
convexAgentRoute as route,
} from "../auth";
export default defineAgent(({ env }) => {
const { AGENT_MODEL_NAME, AGENT_MODEL_PROVIDER } = parseAgentEnv(env);
return {
description:
"Proposes Work Definitions, Design Packets, and structured questions.",
instructions: INSTRUCTIONS,
model: `${AGENT_MODEL_PROVIDER}/${AGENT_MODEL_NAME}`,
thinkingLevel: "medium",
tools: createWorkPlannerTools(env),
};
});
packages/agents/src/agents/zopu.ts
+10 -15
View File
@@ -1,31 +1,26 @@
import { parseAgentEnv } from "@code/env/agent";
import { defineAgent } from "@flue/runtime";
import { local } from "@flue/runtime/node";
import INSTRUCTIONS from "../prompts/zopu-instructions.md" with { type: "markdown" };
import { createCreatePrForIssueTool } from "../tools/create-pr-for-issue";
import { createSliceOneTools } from "../tools/slice-one";
import { createSignalRoutingTools } from "../tools/signal-routing";
import { createWorkCommandTools } from "../tools/work-commands";
export {
convexAgentRoute as attachments,
convexAgentRoute as route,
} from "../auth";
// The host checkout the chat agent explores. The `local()` sandbox reuses the
// machine's existing shell, Git, and Tea install unchanged; its cwd becomes the
// default working directory for every command the agent runs.
const ZOPU_CODE_PATH = "/Users/puter/Workspace/zopu/code";
conversationRoute as attachments,
conversationRoute as route,
} from "../auth/conversation-route";
export default defineAgent(({ env, id }) => {
const { AGENT_MODEL_NAME, AGENT_MODEL_PROVIDER } = parseAgentEnv(env);
return {
description:
"Turns actionable conversation into provenanced Signals and proposed Work.",
"Turns actionable conversation into provenanced Signals, proposed Work, and execution requests.",
instructions: INSTRUCTIONS,
model: `${AGENT_MODEL_PROVIDER}/${AGENT_MODEL_NAME}`,
sandbox: local({ cwd: ZOPU_CODE_PATH }),
thinkingLevel: "high",
tools: [...createSliceOneTools(id, env), createCreatePrForIssueTool(env)],
tools: [
...createSignalRoutingTools(id, env),
...createWorkCommandTools(id, env),
],
};
});
packages/agents/src/app.ts
+18 -20
View File
@@ -5,11 +5,12 @@ import type { Fetchable } from "@flue/runtime/routing";
import { flue } from "@flue/runtime/routing";
import { Hono } from "hono";
import { internalRoute } from "./auth/internal-route";
import { runtimeRegistry } from "./runtime/agent-os-registry";
import {
cancelAgentOsAttempt,
executeAgentOsAttempt,
runtimeRegistry,
} from "./runtime/agent-os";
} from "./runtime/attempt-runner";
const agentEnv = parseAgentEnv(process.env);
@@ -34,12 +35,7 @@ registerProvider(agentEnv.AGENT_MODEL_PROVIDER, {
const app = new Hono();
app.post("/internal/work-attempts/execute", async (context) => {
if (
context.req.header("authorization") !== `Bearer ${agentEnv.FLUE_DB_TOKEN}`
) {
return context.json({ error: "Unauthorized" }, 401);
}
app.post("/internal/work-attempts/execute", internalRoute, async (context) => {
try {
return context.json(await executeAgentOsAttempt(await context.req.json()));
} catch (error) {
@@ -65,19 +61,21 @@ app.post("/internal/work-attempts/execute", async (context) => {
}
});
app.post("/internal/work-attempts/:workspaceKey/cancel", async (context) => {
if (
context.req.header("authorization") !== `Bearer ${agentEnv.FLUE_DB_TOKEN}`
) {
return context.json({ error: "Unauthorized" }, 401);
app.post(
"/internal/work-attempts/:workspaceKey/cancel",
internalRoute,
async (context) => {
const body = (await context.req.json()) as { attemptId?: unknown };
if (typeof body.attemptId !== "string" || body.attemptId.length === 0) {
return context.json({ error: "attemptId is required" }, 400);
}
await cancelAgentOsAttempt(
context.req.param("workspaceKey"),
body.attemptId
);
return context.json({ cancelled: true });
}
const body = (await context.req.json()) as { attemptId?: unknown };
if (typeof body.attemptId !== "string" || body.attemptId.length === 0) {
return context.json({ error: "attemptId is required" }, 400);
}
await cancelAgentOsAttempt(context.req.param("workspaceKey"), body.attemptId);
return context.json({ cancelled: true });
});
);
app.all("/api/rivet/*", (context) => runtimeRegistry.handler(context.req.raw));
packages/agents/src/auth.ts → packages/agents/src/auth/conversation-route.ts
+2 -2
View File
@@ -1,10 +1,10 @@
import { parseAgentEnv } from "@code/env/agent";
import type { AgentRouteHandler } from "@flue/runtime";
import { withTurnAdmission } from "./admission-context";
import { withTurnAdmission } from "../adapters/admission-context";
/** Only Convex may invoke or observe the organization-scoped Flue agent. */
export const convexAgentRoute: AgentRouteHandler = async (context, next) => {
export const conversationRoute: AgentRouteHandler = async (context, next) => {
const env = parseAgentEnv(process.env);
const authorization = context.req.header("authorization");
if (authorization !== `Bearer ${env.FLUE_DB_TOKEN}`) {
packages/agents/src/auth/internal-route.ts
+12
View File
@@ -0,0 +1,12 @@
import { parseAgentEnv } from "@code/env/agent";
import type { MiddlewareHandler } from "hono";
/** Internal service authorization for execution/cancel routes. Not client-facing. */
export const internalRoute: MiddlewareHandler = async (context, next) => {
const env = parseAgentEnv(process.env);
const authorization = context.req.header("authorization");
if (authorization !== `Bearer ${env.FLUE_DB_TOKEN}`) {
return context.json({ error: "Unauthorized" }, 401);
}
return await next();
};
packages/agents/src/auth/workflow-route.ts
+39
View File
@@ -0,0 +1,39 @@
import { parseAgentEnv } from "@code/env/agent";
import type { WorkflowRouteHandler } from "@flue/runtime";
import * as v from "valibot";
const WorkflowInputEnvelope = v.object({
input: v.object({
organizationId: v.string(),
requestId: v.string(),
}),
});
/** Service authorization for finite Flue workflows. No turn headers required. */
export const workflowRoute: WorkflowRouteHandler = async (context, next) => {
const env = parseAgentEnv(process.env);
const authorization = context.req.header("authorization");
if (authorization !== `Bearer ${env.FLUE_DB_TOKEN}`) {
return context.json({ error: "Unauthorized" }, 401);
}
if (context.req.method !== "POST") {
return await next();
}
const body = (await context.req.json().catch(() => null)) as unknown;
const parsed = v.safeParse(WorkflowInputEnvelope, body);
if (!parsed.success) {
return context.json({ error: "Invalid workflow input envelope" }, 400);
}
const organizationId = context.req.header("x-zopu-organization-id");
if (
!organizationId ||
organizationId !== parsed.output.input.organizationId
) {
return context.json({ error: "Forbidden" }, 403);
}
return await next();
};
packages/agents/src/prompts/zopu-instructions.md
+12 -29
View File
@@ -1,4 +1,4 @@
You are Zopu for product Slice 1 and the Work planning handoff.
You are Zopu, the conversational agent that turns user messages into provenanced Signals, proposed Work, and execution requests.
## Your role
@@ -12,15 +12,18 @@ When a user sends a message, follow this decision flow:
Direct questions, casual conversation, and image-reading requests must be answered immediately without calling any tools.
2. **Identify project context.** Call list_projects. If there is one project, use it. If there are several and the request is ambiguous, ask one focused question.
2. **Identify project context.** Call `list_projects`. If there is one project, use it. If there are several and the request is ambiguous, ask one focused question.
3. **Create a Signal (only when actionable).** When the message is actionable: a. Call list_signal_evidence to see the exact admitted user messages. b. Select the message IDs that compose the problem statement. c. Call create_signal with a structured problem statement (title, summary, desiredOutcome, constraints). The problem statement must faithfully represent the user's own intent. Do not invent scope they did not mention. d. Include the projectId when the project is known.
3. **Create a Signal (only when actionable).** a. Call `list_signal_evidence` to see the exact admitted user messages. b. Select the message IDs that compose the problem statement. c. Call `create_signal` with a structured problem statement (title, summary, desiredOutcome, constraints). The problem statement must faithfully represent the user's own intent. Do not invent scope they did not mention. d. Include the `projectId` when the project is known.
4. **Route the Signal.** After creating the Signal: a. Call list_proposed_work for the project. b. If the Signal clearly describes the same desired outcome as existing Work, call attach_signal_to_work. c. Otherwise call create_work_from_signal. e. If genuinely uncertain whether to attach or create, ask one focused question.
4. **Route the Signal.** a. Call `list_proposed_work` for the project. b. If the Signal clearly describes the same desired outcome as existing Work, call `attach_signal_to_work`. c. Otherwise call `create_work_from_signal`. d. If genuinely uncertain whether to attach or create, ask one focused question.
5. **Explain the outcome.** Tell the user clearly what happened:
5. **Execution requests.** When a user explicitly asks to implement or start work on a ready Work unit, call `start_work` with the work ID. The system handles planning, execution, verification, and delivery. Do not claim implementation has started until the tool returns a runId.
6. **Explain the outcome.** Tell the user clearly what happened:
- "Captured [Signal title] and linked it to [Work title]."
- "Captured [Signal title] and proposed [Work title]."
- "Started execution of [Work title]."
- Keep the response brief; the product renders the durable Work card separately.
## Rules
@@ -31,27 +34,7 @@ When a user sends a message, follow this decision flow:
- Ask at most one focused clarification when genuinely ambiguous.
- Preserve project and organization scope at all times.
- Repeated delivery of the same message must not create duplicate Signals or attachments. The backend is idempotent.
- Never claim you created a Signal until the tool call returns successfully.
- Do not start implementation, planning, sandboxes, verification, or delivery. Those are explicitly outside Slice 1.
- After creating proposed Work, the system may invoke the private work-planner. Never claim that a Definition, Design, approval, or implementation exists unless Convex reports it.
- Proposed Work is the only Work status you directly create.
## Issue resolution with create_pr_for_issue
When a user wants an open Gitea issue implemented and turned into a pull request, call `create_pr_for_issue` with the issue number or URL.
- The tool accepts the request immediately and runs the AgentOS implementation, branch push, and PR creation in the background.
- Tell the user the issue was accepted for background implementation; do not claim the PR exists yet and do not wait for completion in the chat turn.
- Do not retry the tool in a tight loop. Background success and failure are written to the agent server logs with the `[create_pr_for_issue]` prefix.
- Do not run repository mutations yourself; the tool owns the implementation and PR pipeline.
## Repository access
Your shell runs inside the `puter/zopu-code` checkout at `/Users/puter/Workspace/zopu/code`, so you can answer questions about this repository and manage its Gitea issues.
- Explore the repository with read-only Git: `git log`, `git status`, `git diff`, `git branch`, `git show`, and reading files. Use this to ground answers about the codebase.
- List open issues with `tea issues list` (defaults to open issues).
- Create a Gitea issue in this repo with `tea issues create --title "<title>" [--description "<details>"]`.
- Never mutate repository state: do not run `git push`, `git merge`, `git rebase`, `git reset`, `git commit`, `git add`, force operations, or any history rewrite.
- Never close, reopen, or edit existing issues (`tea issues close|reopen|edit`); only list and create.
- Never expose credentials, tokens, or the contents of Tea/Git config files.
- Never claim you created a Signal, started Work, or produced a result until the tool call returns successfully.
- Do not perform implementation, planning, design, verification, or delivery yourself. Those are handled by dedicated workflows and the execution runtime.
- You have no repository access. You cannot run shell commands, inspect code, manage Git, or create pull requests. Focus on conversation, Signal capture, and Work orchestration.
- Proposed Work is the only Work status you directly create. The system advances Work through planning (Definition, Design) automatically. Work becomes ready for execution when planning completes.
packages/agents/src/runner.ts
+1 -1
View File
@@ -1,3 +1,3 @@
import { runtimeRegistry } from "./runtime/agent-os";
import { runtimeRegistry } from "./runtime/agent-os-registry";
await runtimeRegistry.startAndWait();
packages/agents/src/runtime/agent-os-registry.ts
+26
View File
@@ -0,0 +1,26 @@
import { makePiAgentOsConfig } from "@code/primitives";
import { agentOS, setup } from "@rivet-dev/agentos";
const MAX_ACP_COMPLETED_MESSAGE_BYTES = 128 * 1024 * 1024;
const piConfig = makePiAgentOsConfig();
/** AgentOS workspace actor. Only the registry runner serves this actor type. */
export const workspace = agentOS<undefined, { token: string }>({
limits: {
acp: {
maxCompletedMessageBytes: MAX_ACP_COMPLETED_MESSAGE_BYTES,
},
},
onBeforeConnect: (_context, params) => {
if (params.token !== process.env.RIVET_WORKSPACE_TOKEN) {
throw new Error("Unauthorized workspace connection");
}
},
options: {
actionTimeout: 10 * 60 * 1000,
},
permissions: piConfig.permissions,
software: piConfig.software,
});
export const runtimeRegistry = setup({ use: { workspace } });
packages/agents/src/runtime/agent-os.ts → packages/agents/src/runtime/attempt-runner.ts
+17 -70
View File
@@ -1,43 +1,20 @@
import { parseAgentEnv } from "@code/env/agent";
import {
makePiAgentOsConfig,
makePiHomeFiles,
decodeWorkAttemptExecutionInput,
WorkAttemptExecutionError,
makePiHomeFiles,
piSessionEnv,
} from "@code/primitives";
import type {
ExecutionEvent,
WorkAttemptExecutionResult,
} from "@code/primitives";
import { agentOS, setup } from "@rivet-dev/agentos";
import { createHostDirBackend } from "@rivet-dev/agentos-core";
import { createClient } from "@rivet-dev/agentos/client";
import { Effect } from "effect";
import { HostRepositoryWorkspace } from "./host-repository";
const MAX_ACP_COMPLETED_MESSAGE_BYTES = 128 * 1024 * 1024;
const piConfig = makePiAgentOsConfig();
const workspace = agentOS<undefined, { token: string }>({
limits: {
acp: {
maxCompletedMessageBytes: MAX_ACP_COMPLETED_MESSAGE_BYTES,
},
},
onBeforeConnect: (_context, params) => {
if (params.token !== process.env.RIVET_WORKSPACE_TOKEN) {
throw new Error("Unauthorized workspace connection");
}
},
options: {
actionTimeout: 10 * 60 * 1000,
},
permissions: piConfig.permissions,
software: piConfig.software,
});
export const runtimeRegistry = setup({ use: { workspace } });
import type { runtimeRegistry } from "./agent-os-registry";
import { classifyRuntimeFailure, executionError } from "./execution-errors";
import { RepositoryWorkspace } from "./repository-workspace";
const event = (
sequence: number,
@@ -52,32 +29,6 @@ const event = (
sequence,
});
const executionError = (
message: string,
reason: WorkAttemptExecutionError["reason"],
retryable: boolean
) => new WorkAttemptExecutionError({ message, reason, retryable });
const classifyRuntimeFailure = (cause: unknown): WorkAttemptExecutionError => {
if (cause instanceof WorkAttemptExecutionError) {
return cause;
}
const message = cause instanceof Error ? cause.message : "Execution failed";
if (/auth|credential|401|403/iu.test(message)) {
return executionError(message, "Authentication", false);
}
if (/clone|checkout|repository|git /iu.test(message)) {
return executionError(message, "RepositoryFailed", false);
}
if (/timeout|timed out/iu.test(message)) {
return executionError(message, "Timeout", true);
}
if (/connect|network|unavailable|502|503|504/iu.test(message)) {
return executionError(message, "ProviderUnavailable", true);
}
return executionError(message, "HarnessFailed", false);
};
export const executeAgentOsAttempt = async (
rawInput: unknown
): Promise<WorkAttemptExecutionResult> => {
@@ -100,9 +51,10 @@ export const executeAgentOsAttempt = async (
workspaceKey: input.workspaceKey,
}),
];
const hostRepository = new HostRepositoryWorkspace();
const prepared = await hostRepository.prepare({
const repository = new RepositoryWorkspace();
const prepared = await repository.prepare({
attemptId: input.attemptId,
baseBranch: input.baseBranch,
cloneUrl: input.repositoryUrl,
credential: input.auth.credential,
piHomeFiles: makePiHomeFiles({
@@ -122,8 +74,8 @@ export const executeAgentOsAttempt = async (
1,
"runtime.preparing",
prepared.created
? "Repository cloned on the execution host"
: "Repository re-cloned"
? "Isolated repository clone created on the execution host"
: "Isolated repository clone recreated"
)
);
const mounts = [
@@ -177,18 +129,15 @@ export const executeAgentOsAttempt = async (
const sessionId = `pi-${input.attemptId}`;
await vm.openSession({
additionalDirectories: [`${prepared.checkoutPath}/.git`],
additionalInstructions:
"Work only inside /workspace/repository. This is an isolated checkout of the project repository. Read AGENTS.md and the relevant product specifications before changing code. Never reveal credentials, push, or open a pull request. Implement the requested change and run focused verification.",
"Work only inside /workspace/repository. This is an isolated clone of the project repository. Read the project documentation and conventions before changing code. Never reveal credentials, push, create branches, or open a pull request. Implement the requested change and run focused verification.",
agent: "pi",
cwd: "/workspace/repository",
env: piSessionEnv(env.AGENT_MODEL_API_KEY),
permissionPolicy: "allow_all",
sessionId,
});
events.push(
event(3, "harness.started", "Pi implementation session started")
);
events.push(event(3, "harness.started", "Implementation session started"));
const promptResult = await vm.prompt({
content: [{ text: input.prompt, type: "text" }],
idempotencyKey: input.attemptId,
@@ -198,19 +147,19 @@ export const executeAgentOsAttempt = async (
const reason =
promptResult.stopReason === "cancelled" ? "Cancelled" : "HarnessFailed";
throw executionError(
`Pi stopped with ${promptResult.stopReason}`,
`Agent stopped with ${promptResult.stopReason}`,
reason,
promptResult.stopReason === "max_tokens" ||
promptResult.stopReason === "max_turn_requests"
);
}
events.push(
event(4, "harness.progress", "Pi implementation turn completed", {
event(4, "harness.progress", "Implementation turn completed", {
stopReason: promptResult.stopReason,
})
);
const collected = await HostRepositoryWorkspace.collect({
const collected = await RepositoryWorkspace.collect({
attemptId: input.attemptId,
baseRevision,
checkoutPath: prepared.checkoutPath,
@@ -221,9 +170,7 @@ export const executeAgentOsAttempt = async (
5,
"repository.changed",
`${changedFiles.length} changed file(s) collected`,
{
candidateRevision,
}
{ candidateRevision }
),
event(6, "runtime.completed", "AgentOS execution completed")
);
@@ -237,8 +184,8 @@ export const executeAgentOsAttempt = async (
events,
summary:
changedFiles.length > 0
? `Pi changed ${changedFiles.length} file(s)`
: "Pi completed without repository changes",
? `Agent changed ${changedFiles.length} file(s)`
: "Agent completed without repository changes",
};
} catch (error) {
throw classifyRuntimeFailure(error);
packages/agents/src/runtime/execution-errors.ts
+29
View File
@@ -0,0 +1,29 @@
import { WorkAttemptExecutionError } from "@code/primitives/execution-runtime";
export const executionError = (
message: string,
reason: WorkAttemptExecutionError["reason"],
retryable: boolean
) => new WorkAttemptExecutionError({ message, reason, retryable });
export const classifyRuntimeFailure = (
cause: unknown
): WorkAttemptExecutionError => {
if (cause instanceof WorkAttemptExecutionError) {
return cause;
}
const message = cause instanceof Error ? cause.message : "Execution failed";
if (/auth|credential|401|403/iu.test(message)) {
return executionError(message, "Authentication", false);
}
if (/clone|checkout|repository|git /iu.test(message)) {
return executionError(message, "RepositoryFailed", false);
}
if (/timeout|timed out/iu.test(message)) {
return executionError(message, "Timeout", true);
}
if (/connect|network|unavailable|502|503|504/iu.test(message)) {
return executionError(message, "ProviderUnavailable", true);
}
return executionError(message, "HarnessFailed", false);
};
packages/agents/src/runtime/host-repository.ts → packages/agents/src/runtime/repository-workspace.ts
+36 -10
View File
@@ -15,6 +15,7 @@ import type { PiHomeFiles } from "@code/primitives";
interface PrepareRepositoryInput {
attemptId: string;
baseBranch?: string;
cloneUrl: string;
/** Provider credential for the initial clone, stripped after cloning. */
credential?: string;
@@ -68,20 +69,40 @@ const changedFilePath = (line: string): string => {
: filePath.slice(renameIndex + renameSeparator.length);
};
// Minimal allow-list so untrusted repo code (postinstall scripts, hooks)
// can never read host secrets like tokens or API keys from the inherited env.
// PATH is needed for git/bun; locale vars prevent encoding issues.
// HOME is intentionally excluded — the runner gets its own /home/zopu mount,
// and forwarding the host HOME would expose host-level dotfiles and credentials.
const SAFE_ENV_KEYS = [
"PATH",
"LANG",
"LC_ALL",
"LC_CTYPE",
"TERM",
"TZ",
] as const;
const safeBaseEnv = (): Record<string, string> => {
const env: Record<string, string> = {};
for (const key of SAFE_ENV_KEYS) {
const value = process.env[key];
if (value !== undefined) {
env[key] = value;
}
}
return env;
};
const runProcess = async (
command: string,
args: readonly string[],
cwd: string,
env: Record<string, string> = {}
): Promise<ProcessResult> => {
const environment = Object.fromEntries(
Object.entries(process.env).filter(
(entry): entry is [string, string] => entry[1] !== undefined
)
);
const child = spawn(command, args, {
cwd,
env: { ...environment, ...env },
env: { ...safeBaseEnv(), ...env },
});
const stderr: Uint8Array[] = [];
const stdout: Uint8Array[] = [];
@@ -107,7 +128,7 @@ const pathExists = async (target: string): Promise<boolean> => {
}
};
export class HostRepositoryWorkspace {
export class RepositoryWorkspace {
readonly #root: string;
readonly #installDependencies: boolean;
@@ -169,12 +190,17 @@ export class HostRepositoryWorkspace {
"Repository clone"
);
} finally {
// The credential env var is never written to disk or git config.
// The askpass script is cleaned up with the workspace.
// The credential only existed on the local cloneEnv object, never on
// process.env. The askpass script is cleaned up with the workspace.
cloneEnv.ZOPU_GIT_TOKEN = "";
delete process.env.ZOPU_GIT_TOKEN;
}
if (input.baseBranch) {
requireSuccess(
await runGit(checkoutPath, ["checkout", input.baseBranch]),
"Base branch checkout"
);
}
const bunExecutable =
process.env.BUN_EXECUTABLE ??
execFileSync("which", ["bun"], { encoding: "utf-8" }).trim();
packages/agents/src/tools/create-pr-for-issue.ts
-572
View File
@@ -1,572 +0,0 @@
import { spawnSync } from "node:child_process";
import type { SpawnSyncOptions } from "node:child_process";
import { createHash, randomUUID } from "node:crypto";
import path from "node:path";
import { parseAgentEnv } from "@code/env/agent";
import { defineTool } from "@flue/runtime";
import type { JsonValue } from "@flue/runtime";
import * as v from "valibot";
import { executeAgentOsAttempt } from "../runtime/agent-os";
/**
* Direct Zopu chat tool that converts a Gitea issue into an AgentOS-implemented
* pull request. It owns the entire issue → harness → push → PR pipeline so the
* chat agent never needs (and must not use) shell mutation for this workflow.
*/
// The host checkout the chat agent explores. Tea and Git resolve this repo's
// Gitea connection; all Tea/Git commands run with this fixed cwd.
const TRUSTED_REPO_PATH = "/Users/puter/Workspace/zopu/code";
const REPO_SLUG = "puter/zopu-code";
// Bounded output so a verbose harness/CLI can never blow up a tool result.
const MAX_OUTPUT = 4000;
const MAX_ISSUE_BODY = 8000;
const MAX_CHANGED_FILES = 200;
const COMMAND_TIMEOUT_MS = 60_000;
const PUSH_TIMEOUT_MS = 120_000;
interface CommandResult {
readonly status: number | null;
readonly stdout: string;
readonly stderr: string;
}
interface IssueDetails {
readonly body: string;
readonly title: string;
readonly url: string;
}
interface ToolFailure {
readonly [key: string]: JsonValue;
readonly error: string;
readonly issue: string;
readonly stage: string;
}
interface ToolSuccess {
readonly [key: string]: JsonValue;
readonly baseBranch: string;
readonly changedFileCount: number;
readonly changedFiles: string[];
readonly headBranch: string;
readonly issue: string;
readonly issueNumber: number;
readonly ok: true;
readonly pullRequestUrl: string;
readonly summary: string;
}
interface ToolAccepted {
readonly [key: string]: JsonValue;
readonly accepted: true;
readonly backgroundId: string;
readonly issue: string;
readonly issueNumber: number;
readonly message: string;
readonly ok: true;
}
const cap = (value: string, limit = MAX_OUTPUT): string =>
value.length <= limit ? value : `${value.slice(0, limit)}…[truncated]`;
// Stable prefix for every background pipeline log line so operators can grep
// for issue→PR outcomes independently of chat traffic.
const LOG_PREFIX = "[create_pr_for_issue]";
/** Writes one stable, prefixed server log line for a background outcome. */
const logBackground = (level: "info" | "error", message: string): void => {
const line = `${LOG_PREFIX} ${message}`;
if (level === "error") {
console.error(line);
} else {
console.log(line);
}
};
/** Logs a structured terminal pipeline result under the stable prefix. */
const logPipelineResult = (
backgroundId: string,
result: ToolFailure | ToolSuccess
): void => {
if ("ok" in result && result.ok === true) {
logBackground(
"info",
`background=${backgroundId} completed issue=#${result.issueNumber} stage=pr url=${result.pullRequestUrl} changedFiles=${result.changedFileCount}`
);
return;
}
const detail =
"reason" in result && typeof result.reason === "string"
? ` reason=${result.reason}`
: "";
logBackground(
"error",
`background=${backgroundId} failed issue=${result.issue} issueNumber=${result.issueNumber} stage=${result.stage}: ${result.error}${detail}`
);
};
/** Last-resort logger for an unexpected throw outside the pipeline's catch. */
const logPipelineError = (backgroundId: string, error: unknown): void => {
const message = error instanceof Error ? error.message : String(error);
logBackground("error", `background=${backgroundId} unexpected: ${message}`);
};
/**
* Runs one command as a direct child process. Arguments are passed to the
* child verbatim — never through a shell — so issue titles, bodies, and
* revisions cannot inject shell metacharacters.
*/
const run = (
command: string,
args: readonly string[],
options: {
cwd?: string;
env?: Record<string, string>;
timeoutMs?: number;
} = {}
): CommandResult => {
const spawnOptions: SpawnSyncOptions = {
cwd: options.cwd ?? TRUSTED_REPO_PATH,
env: { ...process.env, ...options.env },
maxBuffer: 1024 * 1024,
timeout: options.timeoutMs ?? COMMAND_TIMEOUT_MS,
};
const result = spawnSync(command, args, spawnOptions);
return {
status: result.status,
stderr: cap((result.stderr ?? "").toString()),
stdout: cap((result.stdout ?? "").toString()),
};
};
/** Extracts the issue number from an issue number, "#23", or a full issue URL. */
const parseIssueNumber = (input: string): number | null => {
// Prefer a /issues/<n> URL path segment so anchors like #comment-45 do not
// hijack the number.
const pathMatch = input.match(/\/issues\/(?<issueNumber>\d+)(?:[^/\d]|$)/iu);
if (pathMatch?.groups?.issueNumber) {
return Math.trunc(Number(pathMatch.groups.issueNumber));
}
const hashMatch = input.match(/#(?<issueNumber>\d+)\b/u);
if (hashMatch?.groups?.issueNumber) {
return Math.trunc(Number(hashMatch.groups.issueNumber));
}
const standalone = input.trim().match(/^\d+$/u);
if (standalone) {
return Math.trunc(Number(standalone[0]));
}
// Last resort: the trailing integer anywhere in the string.
const value = Math.trunc(Number(input.match(/\d+/gu)?.at(-1)));
if (Number.isFinite(value) && value > 0) {
return value;
}
return null;
};
const readIssue = (issueNumber: number): IssueDetails | ToolFailure => {
const result = run("tea", [
"issues",
String(issueNumber),
"--fields",
"index,title,body,url,state",
"--output",
"json",
]);
if (result.status !== 0) {
return {
error: `tea could not read issue #${issueNumber}.`,
issue: String(issueNumber),
issueNumber,
stage: "read-issue",
stderr: result.stderr,
};
}
let data: {
body?: string;
state?: string;
title?: string;
url?: string;
};
try {
data = JSON.parse(result.stdout) as typeof data;
} catch (error) {
return {
error: `Could not parse the issue payload from tea: ${
error instanceof Error ? error.message : String(error)
}`,
issue: String(issueNumber),
issueNumber,
stage: "read-issue",
stderr: result.stdout,
};
}
const title = (data.title ?? "").trim();
if (title.length === 0) {
return {
error: `Issue #${issueNumber} has no title.`,
issue: String(issueNumber),
issueNumber,
stage: "read-issue",
};
}
return {
body: (data.body ?? "").trim(),
title,
url: (data.url ?? "").trim(),
};
};
const buildPrompt = (issueNumber: number, issue: IssueDetails): string => {
const body =
issue.body.length > MAX_ISSUE_BODY
? `${issue.body.slice(0, MAX_ISSUE_BODY)}…[issue body truncated]`
: issue.body;
return [
`Resolve Gitea issue #${issueNumber} in this repository and ship a complete implementation.`,
"",
`## Issue #${issueNumber}: ${issue.title}`,
"",
body || "_(The issue provided no additional description.)_",
"",
"## Instructions",
"- Read AGENTS.md and the relevant product/architecture specifications before changing code.",
"- Implement exactly the change described by the issue with focused, correct edits.",
"- Reuse existing conventions; do not introduce unrelated scope.",
"- Run focused verification for your change before finishing.",
"- Do NOT push, open a pull request, rewrite history, or modify the read-only base checkout.",
" The orchestrator collects your working-tree changes, pushes a candidate branch, and opens the pull request after you finish.",
].join("\n");
};
/** Mirrors HostRepositoryWorkspace's worktree layout to locate the harness checkout. */
const harnessCheckoutPath = (attemptId: string): string => {
const root = process.env.AGENT_WORKSPACE_ROOT ?? "/var/lib/zopu/workspaces";
const identity = createHash("sha256")
.update(attemptId)
.digest("hex")
.slice(0, 24);
return path.join(root, identity, "repository");
};
const detectBaseBranch = (): string => {
const result = run("git", ["rev-parse", "--abbrev-ref", "origin/HEAD"], {
cwd: TRUSTED_REPO_PATH,
});
if (result.status === 0) {
const ref = result.stdout.trim();
const branch = ref.startsWith("origin/")
? ref.slice("origin/".length)
: ref;
if (branch.length > 0) {
return branch;
}
}
return "master";
};
const pushCandidate = (
checkoutPath: string,
candidateRevision: string,
pushBranch: string
): true | ToolFailure => {
const result = run(
"git",
[
"-C",
checkoutPath,
"push",
"origin",
`${candidateRevision}:refs/heads/${pushBranch}`,
],
{ cwd: checkoutPath, timeoutMs: PUSH_TIMEOUT_MS }
);
if (result.status !== 0) {
return {
error: `Could not push candidate branch "${pushBranch}" from the harness worktree.`,
issue: "",
stage: "push",
stderr: result.stderr,
};
}
return true;
};
const findPullUrl = (head: string): string => {
const result = run("tea", [
"pulls",
"list",
"--state",
"open",
"--fields",
"index,title,head,url",
"--output",
"json",
]);
if (result.status !== 0) {
return "";
}
try {
const pulls = JSON.parse(result.stdout) as {
head?: string;
url?: string;
}[];
const match = pulls.find(
(pull) => pull.head === head || (pull.head ?? "").endsWith(`:${head}`)
);
return (match?.url ?? "").trim();
} catch {
return "";
}
};
const createPullRequest = (
issueNumber: number,
title: string,
head: string,
base: string
): ToolFailure | { url: string } => {
const prTitle = `Fix #${issueNumber}: ${title}`;
const result = run("tea", [
"pulls",
"create",
"--head",
head,
"--base",
base,
"--title",
prTitle,
"--description",
`Fixes #${issueNumber}`,
]);
if (result.status !== 0) {
return {
error: `tea could not create the pull request for branch "${head}".`,
issue: "",
stage: "create-pr",
stderr: result.stderr,
stdout: result.stdout,
};
}
const combined = `${result.stdout}\n${result.stderr}`;
const urlMatch = combined.match(/https?:\/\/[^\s)]+\/pulls\/\d+/u);
if (urlMatch) {
return { url: urlMatch[0] };
}
// Fall back to listing open pulls and matching the head branch.
const fallback = findPullUrl(head);
if (fallback.length > 0) {
return { url: fallback };
}
return {
error:
"The pull request was created, but its URL could not be captured from tea. Inspect open pulls in Gitea.",
issue: "",
stage: "create-pr",
stdout: result.stdout,
};
};
/**
* Runs the full issue → AgentOS → push → PR pipeline and returns a structured
* terminal result. Used both by the background scheduler (chat path) and any
* direct/internal caller that wants the structured outcome. Never rejects: every
* failure is caught and returned as a {@link ToolFailure}.
*/
const runIssuePrPipeline = async (params: {
readonly giteaUrl: string;
readonly giteaToken: string | undefined;
readonly issue: string;
readonly issueNumber: number;
readonly identifiers: {
readonly attemptId: string;
readonly headBranch: string;
readonly runId: string;
readonly workspaceKey: string;
};
}): Promise<ToolFailure | ToolSuccess> => {
const { giteaUrl, giteaToken, issue, issueNumber, identifiers } = params;
try {
const issueOrFailure = readIssue(issueNumber);
if ("error" in issueOrFailure) {
return issueOrFailure;
}
const issueDetails = issueOrFailure;
const baseBranch = detectBaseBranch();
const prompt = buildPrompt(issueNumber, issueDetails);
// The fixed-checkout harness path ignores `auth`/`repositoryUrl` for
// cloning (it always uses the mounted Zopu source worktree); they are
// still supplied as valid schema values. GITEA_TOKEN is optional.
const result = await executeAgentOsAttempt({
attemptId: identifiers.attemptId,
auth: {
credential: giteaToken ?? "unused-by-fixed-checkout",
provider: "gitea",
serverUrl: giteaUrl,
},
baseBranch,
prompt,
repositoryUrl: `${giteaUrl}/${REPO_SLUG}`,
runId: identifiers.runId,
workId: `issue-${issueNumber}`,
workspaceKey: identifiers.workspaceKey,
});
const changedFiles = result.changedFiles.slice(0, MAX_CHANGED_FILES);
if (
result.changedFiles.length === 0 ||
result.candidateRevision === result.baseRevision
) {
return {
error:
"The AgentOS session completed without producing any repository changes, so there is nothing to turn into a pull request.",
issue,
issueNumber,
stage: "harness",
summary: result.summary,
};
}
const checkoutPath = harnessCheckoutPath(identifiers.attemptId);
const pushed = pushCandidate(
checkoutPath,
result.candidateRevision,
identifiers.headBranch
);
if (pushed !== true) {
return {
...pushed,
attemptId: identifiers.attemptId,
baseRevision: result.baseRevision,
candidateRevision: result.candidateRevision,
headBranch: identifiers.headBranch,
issue,
issueNumber,
workspaceKey: identifiers.workspaceKey,
};
}
const pr = createPullRequest(
issueNumber,
issueDetails.title,
identifiers.headBranch,
baseBranch
);
if ("error" in pr) {
return {
...pr,
attemptId: identifiers.attemptId,
baseRevision: result.baseRevision,
candidateRevision: result.candidateRevision,
headBranch: identifiers.headBranch,
issue,
issueNumber,
workspaceKey: identifiers.workspaceKey,
};
}
return {
baseBranch,
changedFileCount: result.changedFiles.length,
changedFiles,
headBranch: identifiers.headBranch,
issue,
issueNumber,
ok: true,
pullRequestUrl: pr.url,
summary: result.summary,
};
} catch (error) {
const message =
error instanceof Error ? error.message : "Unexpected failure";
const reason =
error !== null &&
typeof error === "object" &&
"reason" in error &&
typeof (error as { reason: unknown }).reason === "string"
? (error as { reason: string }).reason
: undefined;
return {
error: message,
issue,
issueNumber,
...(reason ? { reason } : {}),
stage: "harness",
};
}
};
export const createCreatePrForIssueTool = (
runtimeEnv: Record<string, string | undefined>
) => {
const agentEnv = parseAgentEnv(runtimeEnv);
const giteaUrl = agentEnv.GITEA_URL;
return defineTool({
description:
'Validate a Gitea issue number or URL immediately, then return an accepted acknowledgement right away while an AgentOS coding session implements the issue, pushes a candidate branch, and opens a PR titled "Fix #N: <title>" with body "Fixes #N" in the background. Pass a Gitea issue number or full issue URL. The created PR URL and any failure are logged to the server log with the "[create_pr_for_issue]" prefix rather than returned to chat; malformed input returns a parse failure immediately.',
input: v.object({
issue: v.pipe(
v.string(),
v.description(
"A Gitea issue number like 23, or a full issue URL such as https://git.openputer.com/puter/zopu-code/issues/23."
)
),
}),
name: "create_pr_for_issue",
run({ input }): ToolFailure | ToolAccepted {
const issueRef = input.issue.trim();
const issueNumber = parseIssueNumber(issueRef);
if (issueNumber === null) {
return {
error: `Could not parse an issue number from "${input.issue}". Provide a Gitea issue number or URL.`,
issue: input.issue,
stage: "parse",
};
}
// Background identifiers are generated up front so the accepted
// response can reference them and the pipeline can reuse them.
const backgroundId = randomUUID();
const attemptId = `create-pr-${issueNumber}-${randomUUID()}`;
const workspaceKey = `issue-${issueNumber}-pr-${randomUUID()}`;
const runId = `issue-pr-${randomUUID()}`;
const shortId = randomUUID().slice(0, 8);
const headBranch = `zopu/issue-${issueNumber}-${shortId}`;
// Schedule the full issue → AgentOS → push → PR pipeline on the next
// microtask boundary so this turn returns immediately. The pipeline
// catches its own failures and logs a stable prefixed line; the IIFE
// never rejects, so the originating chat turn cannot fail.
queueMicrotask(() => {
void (async () => {
try {
const result = await runIssuePrPipeline({
giteaToken: runtimeEnv.GITEA_TOKEN,
giteaUrl,
identifiers: { attemptId, headBranch, runId, workspaceKey },
issue: input.issue,
issueNumber,
});
logPipelineResult(backgroundId, result);
} catch (error) {
logPipelineError(backgroundId, error);
}
})();
});
return {
accepted: true,
backgroundId,
issue: input.issue,
issueNumber,
message: `Issue #${issueNumber} accepted. An AgentOS session will implement it, push branch "${headBranch}", and open a pull request in the background. The created PR URL and any failure are logged with the "${LOG_PREFIX}" prefix.`,
ok: true,
};
},
});
};
packages/agents/src/tools/slice-one.ts → packages/agents/src/tools/signal-routing.ts
+4 -6
View File
@@ -1,9 +1,9 @@
import { parseAgentEnv } from "@code/env/agent";
import { defineTool } from "@flue/runtime";
import { ConvexHttpClient } from "convex/browser";
import { makeFunctionReference } from "convex/server";
import * as v from "valibot";
import { createConvexServiceClient } from "../adapters/convex-client";
const listProjectsRef = makeFunctionReference<
"query",
{ organizationId: string; token: string },
@@ -52,13 +52,11 @@ const attachSignalRef = makeFunctionReference<
{ attached: boolean; workId: string }
>("works:attachSignalToWork");
export const createSliceOneTools = (
export const createSignalRoutingTools = (
organizationId: string,
runtimeEnv: Record<string, string | undefined>
) => {
const env = parseAgentEnv(runtimeEnv);
const client = new ConvexHttpClient(env.CONVEX_URL);
const token = env.FLUE_DB_TOKEN;
const { client, token } = createConvexServiceClient(runtimeEnv);
return [
defineTool({
packages/agents/src/tools/work-commands.ts
+79
View File
@@ -0,0 +1,79 @@
import { defineTool } from "@flue/runtime";
import { makeFunctionReference } from "convex/server";
import * as v from "valibot";
import { createConvexServiceClient } from "../adapters/convex-client";
const startWorkRef = makeFunctionReference<
"mutation",
{
organizationId: string;
sliceId?: string;
token: string;
workId: string;
},
{ runId: string }
>("workExecution:start");
const cancelWorkRef = makeFunctionReference<
"mutation",
{ organizationId: string; runId: string; token: string },
{ cancelled: boolean }
>("workExecution:cancel");
const retryWorkRef = makeFunctionReference<
"mutation",
{ organizationId: string; runId: string; token: string },
{ runId: string }
>("workExecution:retry");
export const createWorkCommandTools = (
organizationId: string,
runtimeEnv: Record<string, string | undefined>
) => {
const { client, token } = createConvexServiceClient(runtimeEnv);
return [
defineTool({
description:
"Start execution of a ready Work unit. Returns the runId. Use only when the user explicitly requests implementation.",
input: v.object({
sliceId: v.optional(v.string()),
workId: v.string(),
}),
name: "start_work",
async run({ input }) {
return await client.mutation(startWorkRef, {
organizationId,
sliceId: input.sliceId,
token,
workId: input.workId,
});
},
}),
defineTool({
description: "Cancel a running Work execution by runId.",
input: v.object({ runId: v.string() }),
name: "cancel_work",
async run({ input }) {
return await client.mutation(cancelWorkRef, {
organizationId,
runId: input.runId,
token,
});
},
}),
defineTool({
description: "Retry a failed or cancelled Work execution by runId.",
input: v.object({ runId: v.string() }),
name: "retry_work",
async run({ input }) {
return await client.mutation(retryWorkRef, {
organizationId,
runId: input.runId,
token,
});
},
}),
];
};
packages/agents/src/tools/work-planner.ts
-69
View File
@@ -1,69 +0,0 @@
import { parseAgentEnv } from "@code/env/agent";
import { defineTool } from "@flue/runtime";
import { ConvexHttpClient } from "convex/browser";
import { makeFunctionReference } from "convex/server";
import * as v from "valibot";
const submitDefinitionRef = makeFunctionReference<
"mutation",
{ workId: string; payloadJson: string; token: string },
{ version: number }
>("workPlanning:submitDefinitionProposal");
const submitDesignRef = makeFunctionReference<
"mutation",
{ workId: string; payloadJson: string; token: string },
{ version: number }
>("workPlanning:submitDesignProposal");
const submitQuestionRef = makeFunctionReference<
"mutation",
{ workId: string; questionJson: string; token: string },
{ accepted: boolean }
>("workPlanning:submitQuestion");
export const createWorkPlannerTools = (
runtimeEnv: Record<string, string | undefined>
) => {
const env = parseAgentEnv(runtimeEnv);
const client = new ConvexHttpClient(env.CONVEX_URL);
const token = env.FLUE_DB_TOKEN;
return [
defineTool({
description:
"Submit a versioned Work Definition proposal. This never approves the Definition.",
input: v.object({ definitionJson: v.string(), workId: v.string() }),
name: "submit_definition_proposal",
async run({ input }) {
return await client.mutation(submitDefinitionRef, {
payloadJson: input.definitionJson,
token,
workId: input.workId,
});
},
}),
defineTool({
description:
"Submit a Design Packet with observable independently verifiable slices. This never approves the Design.",
input: v.object({ designJson: v.string(), workId: v.string() }),
name: "submit_design_proposal",
async run({ input }) {
return await client.mutation(submitDesignRef, {
payloadJson: input.designJson,
token,
workId: input.workId,
});
},
}),
defineTool({
description: "Submit one structured question for human attention.",
input: v.object({ questionJson: v.string(), workId: v.string() }),
name: "submit_question",
async run({ input }) {
return await client.mutation(submitQuestionRef, {
questionJson: input.questionJson,
token,
workId: input.workId,
});
},
}),
];
};
packages/agents/src/workflows/plan-work.ts
+121
View File
@@ -0,0 +1,121 @@
import { parseAgentEnv } from "@code/env/agent";
import { defineAgent, defineWorkflow } from "@flue/runtime";
import * as v from "valibot";
export { workflowRoute as route } from "../auth/workflow-route";
// --- Input ---
const questionSchema = v.object({
alternatives: v.array(v.string()),
id: v.string(),
impact: v.picklist(["low", "medium", "high"]),
prompt: v.string(),
recommendation: v.optional(v.string()),
});
const planWorkInput = v.object({
context: v.object({
existingDefinition: v.optional(v.string()),
problemStatement: v.string(),
}),
organizationId: v.string(),
phase: v.picklist(["definition", "design"]),
requestId: v.string(),
workId: v.string(),
});
// --- Output ---
const definitionSchema = v.object({
acceptanceCriteria: v.array(v.string()),
affectedUsers: v.array(v.string()),
assumptions: v.array(v.string()),
constraints: v.array(v.string()),
desiredOutcome: v.string(),
inScope: v.array(v.string()),
outOfScope: v.array(v.string()),
problem: v.string(),
requiredArtifacts: v.array(v.string()),
risk: v.picklist(["low", "medium", "high"]),
});
const sliceSchema = v.object({
codeBoundaries: v.array(v.string()),
dependsOn: v.array(v.string()),
evidenceRequirements: v.array(v.string()),
id: v.string(),
objective: v.string(),
observableBehavior: v.string(),
reviewRequired: v.boolean(),
title: v.string(),
verification: v.array(v.string()),
});
const designSchema = v.object({
architectureSummary: v.string(),
callFlowDelta: v.array(v.string()),
concerns: v.array(v.string()),
evidenceRequirements: v.array(v.string()),
fileTreeDelta: v.array(v.string()),
invariants: v.array(v.string()),
keyInterfaces: v.array(v.string()),
slices: v.array(sliceSchema),
tradeoffs: v.array(v.string()),
});
const planWorkOutput = v.union([
v.object({
definition: definitionSchema,
phase: v.literal("definition"),
questions: v.array(questionSchema),
}),
v.object({
design: designSchema,
phase: v.literal("design"),
questions: v.array(questionSchema),
}),
]);
// --- Agent ---
const planWorkAgent = defineAgent(({ env }) => {
const { AGENT_MODEL_NAME, AGENT_MODEL_PROVIDER } = parseAgentEnv(env);
return {
description:
"Produces a Work Definition or Design Packet with structured questions.",
instructions: `You are Zopu's Work Planner. You receive a Work problem statement and produce either a Definition or a Design Packet depending on the requested phase.
For the definition phase, produce a precise problem statement, desired outcome, acceptance criteria, scope boundaries, assumptions, constraints, and risk level. Ask focused high-impact questions only when a decision genuinely blocks the work.
For the design phase, produce an architecture summary, file tree delta, call flow delta, key interfaces, invariants, evidence requirements, 1-4 observable and independently verifiable slices, tradeoffs, and concerns. Each slice must have a clear objective, observable behavior, verification steps, and code boundaries. Dependencies must reference earlier slices.
Never approve, execute, or claim implementation. Your output is a proposal that Convex validates and persists.`,
model: `${AGENT_MODEL_PROVIDER}/${AGENT_MODEL_NAME}`,
thinkingLevel: "high",
};
});
// --- Workflow ---
export default defineWorkflow({
agent: planWorkAgent,
input: planWorkInput,
output: planWorkOutput,
async run({ input, harness }) {
const session = await harness.session();
const instruction =
input.phase === "definition"
? `Produce a Work Definition for the following problem:\n\n${input.context.problemStatement}`
: `Produce a Design Packet based on this approved Definition:\n\n${input.context.existingDefinition ?? input.context.problemStatement}`;
const response = await session.prompt(instruction, {
result: planWorkOutput,
});
if (response.data.phase !== input.phase) {
throw new Error(
`Planner returned phase "${response.data.phase}" but "${input.phase}" was requested`
);
}
return response.data;
},
});