feat: ship agent onboarding runtime

This commit is contained in:
-Puter
2026-08-04 01:38:48 +05:30
parent 48205a5e19
commit b02200b3dc
68 changed files with 10522 additions and 4156 deletions
+17
View File
@@ -0,0 +1,17 @@
import { defineConfig } from "@flue/runtime/config";
// Flue 2.0 project configuration.
// - target: 'node' builds a long-running Node HTTP server.
// - app: the route map (src/app.ts) — health + private internal routes + agent router.
// - db: file-backed SQLite persistence (src/db.ts) for canonical conversation state.
// - agents: scoped to src/agents/ for the 'use agent' scan.
export default defineConfig({
agents: "agents/**/*.ts",
app: "src/app.ts",
db: "src/db.ts",
// Custom providers are registered at runtime via setProvider() in app.ts.
// Empty list = register none from the built-in catalog; the Cheaptricks
// provider is wired programmatically.
providers: [],
target: "node",
});
+39
View File
@@ -0,0 +1,39 @@
{
"name": "@code/agents",
"version": "0.0.0",
"private": true,
"type": "module",
"exports": {
".": "./src/index.ts"
},
"scripts": {
"dev": "vite dev",
"build": "vite build",
"preview": "vite preview",
"check-types": "tsc --noEmit",
"start": "node dist/server.mjs"
},
"dependencies": {
"@agentos-software/common": "catalog:",
"@agentos-software/git": "catalog:",
"@code/env": "workspace:*",
"@earendil-works/pi-ai": "catalog:",
"@flue/runtime": "catalog:",
"@flue/sdk": "catalog:",
"@hono/node-server": "catalog:",
"@rivet-dev/agentos": "catalog:",
"convex": "catalog:",
"hono": "catalog:",
"rivetkit": "catalog:",
"valibot": "^1.0.0",
"zod": "catalog:"
},
"devDependencies": {
"@code/config": "workspace:*",
"@flue/cli": "catalog:",
"@flue/vite": "catalog:",
"@types/node": "catalog:",
"typescript": "catalog:",
"vite": "catalog:"
}
}
+469
View File
@@ -0,0 +1,469 @@
import path from "node:path";
import { env } from "@code/env/agent";
import type { CodeExecutionResult } from "@rivet-dev/agentos";
import { createClient } from "@rivet-dev/agentos/client";
import type { registry } from "../runner";
// AgentOS adapter: resolves one project workspace actor keyed structurally
// by projectId, and exposes only safe read-oriented operations for the
// onboarding runtime endpoint and the conversation agent's tools.
//
// The adapter talks to the durable AgentOS RivetKit actor registered in this
// same process (src/runner.ts, served in serverless mode via the Hono app).
// It caches only the lightweight actor client, never VM handles — so it
// survives engine restarts and process recycling. Each operation re-resolves
// the actor handle via getOrCreate + resolve, which is stateless and safe
// across lifecycle boundaries.
//
// It does NOT use Pi for onboarding — Pi capability is preserved only for
// later GLM-5.2 work, and this module never invokes it.
/**
* Repository path inside a project VM.
* All repos clone to /workspace/repo inside the VM.
*/
export const REPO_PATH = "/workspace/repo";
export interface ProjectVm {
vmId: string;
workspacePath: string;
repositoryPath: string;
}
export interface DirListEntry {
name: string;
kind: "file" | "directory" | "symlink";
size: number;
}
export interface RepoDescription {
commit: string;
rootEntries: DirListEntry[];
}
export interface SearchResult {
path: string;
snippet: string;
}
// ---------------------------------------------------------------------------
// Safe-command policy for execSafe.
// Only static inspected command names are permitted; shell strings are
// rejected. Arguments are passed as an array (no shell interpolation).
// ---------------------------------------------------------------------------
const ALLOWED_COMMANDS: ReadonlyMap<string, readonly string[]> = new Map([
// Git read-only.
[
"git",
[
"rev-parse",
"log",
"status",
"ls-files",
"remote",
"branch",
"describe",
"show",
],
],
// Text inspection.
["grep", ["-r", "-rn", "-i", "--include", "-l"]],
["find", ["-name", "-type", "-maxdepth", "-mindepth"]],
["wc", ["-l", "-w", "-c"]],
["head", ["-n", "-c"]],
["tail", ["-n", "-c"]],
["cat", []],
["ls", ["-la", "-l", "-a"]],
]);
/**
* Validate that a command is statically permitted and that its args are a
* safe list (no shell metacharacters, no strings). This is the only gate
* between the model and process execution.
*/
export const validateSafeCommand = (
command: string,
args: readonly string[]
): void => {
if (typeof command !== "string" || command.length === 0) {
throw new Error("execSafe requires a non-empty command name");
}
// Reject anything that looks like a shell string (spaces, pipes, redirects).
if (/\s|[;&|<>`$()]/u.test(command)) {
throw new Error(`execSafe rejects shell expression: ${command}`);
}
// Resolve the command to its base name (no path traversal through /bin/...).
const base = path.basename(command);
const allowedFlags = ALLOWED_COMMANDS.get(base);
if (!allowedFlags) {
throw new Error(`execSafe command not permitted: ${base}`);
}
for (const arg of args) {
if (typeof arg !== "string") {
throw new TypeError("execSafe args must all be strings");
}
// Reject shell metacharacters in arguments.
if (/[;&|<>`$()]/u.test(arg)) {
throw new Error(`execSafe arg contains shell metacharacters: ${arg}`);
}
// Flags must be in the allowlist (starts with -). Non-flag args (paths,
// patterns, numbers) pass through — they cannot invoke a shell.
if (
arg.startsWith("-") &&
!allowedFlags.includes(arg) &&
!arg.startsWith("--include=")
) {
throw new Error(`execSafe flag not permitted for ${base}: ${arg}`);
}
}
};
/**
* Prevent path traversal: resolve a path against the repository root and
* ensure the result stays inside it. Rejects absolute paths escaping the
* root and ../ sequences.
*/
export const safeRepoPath = (
repoRoot: string,
relativePath: string
): string => {
// Normalize and strip leading slashes so "/etc/passwd" can't escape.
const cleaned = relativePath.replace(/^\/+/u, "");
const resolved = path.resolve(repoRoot, cleaned);
const normalizedRoot = path.resolve(repoRoot);
if (
resolved !== normalizedRoot &&
!resolved.startsWith(normalizedRoot + path.sep)
) {
throw new Error(
`Path traversal rejected: ${relativePath} escapes repository root`
);
}
return resolved;
};
// ---------------------------------------------------------------------------
// Actor client.
//
// The client is the only durable handle the adapter caches. Actor (VM)
// handles are resolved per-operation through getOrCreate + resolve, so the
// adapter is stateless with respect to VM lifecycle and survives process
// recycling.
// ---------------------------------------------------------------------------
/**
* The public actor client surface used by this adapter. The actor handle is
* stateless: every action resolves it against the engine as needed.
*/
interface ProjectWorkspaceClient {
projectWorkspace: {
getOrCreate: (
key: readonly [projectId: string],
options: { params: { token: string } }
) => {
resolve: () => Promise<string>;
process: {
execFile: (
command: string,
args?: readonly string[]
) => Promise<CodeExecutionResult>;
};
filesystem: {
exists: (target: string) => Promise<boolean>;
readdirEntries: (target: string) => Promise<readonly DirStatEntry[]>;
readFile: (target: string) => Promise<Uint8Array>;
mkdir: (
target: string,
options?: { recursive?: boolean }
) => Promise<void>;
writeFile: (
target: string,
content: string | Uint8Array
) => Promise<void>;
};
};
};
}
/** Entry returned by filesystem.readdirEntries. */
interface DirStatEntry {
name: string;
isDirectory: boolean;
isSymbolicLink: boolean;
size: number;
}
/**
* Lazily-created actor client. Cached for the process lifetime; the client
* object is safe to reuse, unlike VM handles which are lifecycle-bound.
*/
let clientCache: ProjectWorkspaceClient | undefined;
const getClient = (): ProjectWorkspaceClient => {
if (!clientCache) {
clientCache = createClient<typeof registry>({
disableMetadataLookup: true,
encoding: "cbor",
endpoint: env.RIVET_ENDPOINT,
}) as unknown as ProjectWorkspaceClient;
}
return clientCache;
};
/**
* Resolve the durable actor handle for a project. The actor is keyed
* structurally by [projectId] (project ids are globally unique). The handle
* is stateless — it re-resolves across lifecycle boundaries — so it is never
* cached.
*/
const getProjectHandle = (projectId: string) =>
getClient().projectWorkspace.getOrCreate([projectId], {
params: { token: env.RIVET_WORKSPACE_TOKEN },
});
/**
* Throw a descriptive Error when an execFile result did not succeed.
* Surfaces the operation, outcome, and available stdout/stderr/error.
*/
const requireExecSuccess = (
result: CodeExecutionResult,
operation: string
): void => {
if (result.outcome !== "succeeded") {
const detail = [result.stderr, result.stdout, result.error.message]
.filter(
(part): part is string => typeof part === "string" && part.length > 0
)
.join(" | ");
throw new Error(
`${operation} failed (outcome=${result.outcome})${detail ? `: ${detail}` : ""}`
);
}
};
/**
* AgentOS owns the actor filesystem. The repository has no host workspace
* projection, so persist its actual in-VM path rather than an invented path.
*/
const workspacePath = (): string => REPO_PATH;
/**
* Get or create a project workspace VM. Idempotent by projectId — the actor
* is keyed structurally by [projectId] and resolved on demand.
*
* The vmId/runtimeId is the actor's resolved id, which remains stable across
* re-resolution.
*/
export const getOrCreateProjectVm = async (
projectId: string
): Promise<ProjectVm> => {
const handle = getProjectHandle(projectId);
const vmId = await handle.resolve();
return {
repositoryPath: REPO_PATH,
vmId,
workspacePath: workspacePath(),
};
};
/**
* Read the current HEAD commit. Used for readability verification and
* ProjectRuntime.repositoryCommit.
*/
export const readRepositoryHead = async (
projectId: string
): Promise<{ commit: string }> => {
const handle = getProjectHandle(projectId);
const result = await handle.process.execFile("git", [
"-C",
REPO_PATH,
"rev-parse",
"HEAD",
]);
requireExecSuccess(result, "git rev-parse HEAD");
return { commit: (result.stdout ?? "").trim() };
};
/**
* Depth-1 branch clone into the project VM. Idempotent: if the repository
* directory already exists, the clone is skipped (assuming a prior successful
* clone). Credentials are never exposed to the model.
*/
export const cloneRepository = async (input: {
projectId: string;
repositoryUrl: string;
branch: string;
}): Promise<{ commit: string }> => {
const handle = getProjectHandle(input.projectId);
// Idempotent: skip if repo dir already exists.
const exists = await handle.filesystem.exists(REPO_PATH).catch(() => false);
if (!exists) {
const result = await handle.process.execFile("git", [
"clone",
"--depth=1",
"--single-branch",
"--branch",
input.branch,
input.repositoryUrl,
REPO_PATH,
]);
requireExecSuccess(result, "git clone");
}
return readRepositoryHead(input.projectId);
};
/**
* List files in a directory relative to the repository root.
* Path traversal is prevented by safeRepoPath().
*/
export const listFiles = async (
projectId: string,
relativePath: string
): Promise<DirListEntry[]> => {
const handle = getProjectHandle(projectId);
const target = safeRepoPath(REPO_PATH, relativePath);
const entries = await handle.filesystem.readdirEntries(target);
return entries.map((e) => {
let kind: DirListEntry["kind"] = "file";
if (e.isDirectory) {
kind = "directory";
} else if (e.isSymbolicLink) {
kind = "symlink";
}
return { kind, name: e.name, size: e.size };
});
};
/**
* Verify repository readability: HEAD commit + root listing + read one file.
*/
export const describeRepository = async (
projectId: string
): Promise<RepoDescription> => {
const { commit } = await readRepositoryHead(projectId);
const entries = await listFiles(projectId, "");
return { commit, rootEntries: entries };
};
/**
* Read a file relative to the repository root. Path traversal prevented.
*/
export const readFile = async (
projectId: string,
relativePath: string
): Promise<{ content: string }> => {
const handle = getProjectHandle(projectId);
const target = safeRepoPath(REPO_PATH, relativePath);
const bytes = await handle.filesystem.readFile(target);
return { content: Buffer.from(bytes).toString("utf-8") };
};
/**
* Search repository files using grep. Returns path + snippet matches.
* Zero matches return an empty array rather than throwing.
*/
export const searchRepository = async (
projectId: string,
query: string
): Promise<{ matches: SearchResult[] }> => {
if (typeof query !== "string" || query.length === 0) {
throw new Error("searchRepository requires a non-empty query string");
}
// Validate exactly the args we execute.
const args = ["-rn", "--include=*", query, REPO_PATH];
validateSafeCommand("grep", args);
const handle = getProjectHandle(projectId);
const result = await handle.process.execFile("grep", args);
if (result.outcome !== "succeeded") {
if (result.exitCode === 1) {
return { matches: [] };
}
requireExecSuccess(result, "grep repository");
}
const lines = (result.stdout ?? "").split("\n").filter(Boolean).slice(0, 50);
return {
matches: lines.map((line) => {
const colonIdx = line.indexOf(":");
const secondColon = line.indexOf(":", colonIdx + 1);
return {
path: line.slice(0, colonIdx),
snippet: line.slice(secondColon + 1, secondColon + 1 + 200),
};
}),
};
};
/**
* Execute a validated safe command in the project VM. Only statically
* inspected commands / safe argument lists are permitted.
*/
export const execSafe = async (
projectId: string,
command: string,
args: readonly string[]
): Promise<{ stdout: string; stderr: string }> => {
validateSafeCommand(command, args);
const handle = getProjectHandle(projectId);
const result = await handle.process.execFile(command, [...args]);
requireExecSuccess(result, `execSafe ${command}`);
return { stderr: result.stderr ?? "", stdout: result.stdout ?? "" };
};
/**
* Scan root .env.example for variable names only (values are never stored
* or displayed). Non-blocking: returns empty array if absent.
*/
export const scanEnvExample = async (
projectId: string
): Promise<{ names: string[] }> => {
const handle = getProjectHandle(projectId);
const envPath = safeRepoPath(REPO_PATH, ".env.example");
const exists = await handle.filesystem.exists(envPath).catch(() => false);
if (!exists) {
return { names: [] };
}
const bytes = await handle.filesystem
.readFile(envPath)
.catch(() => new Uint8Array());
const content = Buffer.from(bytes).toString("utf-8");
const names: string[] = [];
for (const line of content.split("\n")) {
const trimmed = line.trim();
if (!trimmed || trimmed.startsWith("#")) {
continue;
}
const [rawName] = trimmed.split("=");
const name = rawName?.trim() ?? "";
if (name && /^[A-Z_][A-Z0-9_]*$/iu.test(name)) {
names.push(name);
}
}
return { names };
};
/**
* Write a generated file under .zopu/artifacts/. This is the only write path
* exposed — it is constrained to the .zopu output directory and never touches
* repository source files.
*/
export const writeArtifactFile = async (
projectId: string,
buildId: string,
filename: string,
content: string
): Promise<{ path: string }> => {
const handle = getProjectHandle(projectId);
const artifactPath = `/workspace/.zopu/artifacts/${buildId}/${filename}`;
// Ensure parent directory exists.
await handle.filesystem.mkdir(`/workspace/.zopu/artifacts/${buildId}`, {
recursive: true,
});
await handle.filesystem.writeFile(artifactPath, content);
return { path: artifactPath };
};
@@ -0,0 +1,103 @@
// Static HTML artifact renderer.
// Takes a typed ArtifactManifest and produces a safe, static HTML page.
// No user-supplied HTML is passed through — all content is escaped and
// rendered through the kit's component templates.
export interface ArtifactManifestSection {
title: string;
body: string;
items?: string[];
}
export interface ArtifactManifest {
kind: string;
title: string;
subtitle?: string;
sections: ArtifactManifestSection[];
generatedAt: string;
projectName?: string;
repositoryCommit?: string;
}
/** Escape HTML special characters to prevent injection. */
const escapeHtml = (text: string): string =>
text
.replaceAll("&", "&amp;")
.replaceAll("<", "&lt;")
.replaceAll(">", "&gt;")
.replaceAll('"', "&quot;")
.replaceAll("'", "&#39;");
const renderSection = (section: ArtifactManifestSection): string => {
const items = section.items
? `<ul class="items">${section.items.map((i) => `<li>${escapeHtml(i)}</li>`).join("")}</ul>`
: "";
return ` <section>
<h2>${escapeHtml(section.title)}</h2>
<p>${escapeHtml(section.body)}</p>
${items}
</section>`;
};
/**
* Render a typed manifest to a polished, static HTML page.
* All content is escaped — the renderer supplies visual quality and safety.
*/
export const renderStaticHtml = (manifest: ArtifactManifest): string => {
const sections = manifest.sections.map(renderSection).join("\n");
const subtitle = manifest.subtitle
? `<p class="subtitle">${escapeHtml(manifest.subtitle)}</p>`
: "";
const meta: string[] = [];
if (manifest.projectName) {
meta.push(`<span>Project: ${escapeHtml(manifest.projectName)}</span>`);
}
if (manifest.repositoryCommit) {
meta.push(
`<span>Commit: <code>${escapeHtml(manifest.repositoryCommit)}</code></span>`
);
}
const metaHtml = meta.length
? `<div class="meta">${meta.join(" ")}</div>`
: "";
return `<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>${escapeHtml(manifest.title)}</title>
<style>
:root { --bg: #0d1117; --card: #161b22; --border: #30363d; --text: #e6edf3; --muted: #8b949e; --accent: #2f81f7; }
* { box-sizing: border-box; margin: 0; padding: 0; }
body { font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', sans-serif; background: var(--bg); color: var(--text); line-height: 1.6; padding: 2rem 1rem; }
.container { max-width: 720px; margin: 0 auto; }
header { margin-bottom: 2rem; }
h1 { font-size: 1.75rem; font-weight: 700; margin-bottom: 0.5rem; }
h2 { font-size: 1.25rem; font-weight: 600; margin-bottom: 0.5rem; color: var(--accent); }
.subtitle { color: var(--muted); font-size: 1rem; }
.meta { display: flex; flex-wrap: wrap; gap: 1rem; margin-top: 0.75rem; font-size: 0.875rem; color: var(--muted); }
.meta code { background: var(--card); padding: 0.1rem 0.3rem; border-radius: 3px; font-size: 0.8rem; }
section { background: var(--card); border: 1px solid var(--border); border-radius: 8px; padding: 1.5rem; margin-bottom: 1rem; }
section p { color: var(--text); margin-bottom: 0.75rem; }
.items { list-style: none; }
.items li { padding: 0.25rem 0 0.25rem 1.25rem; position: relative; color: var(--muted); }
.items li::before { content: "\\2022"; position: absolute; left: 0; color: var(--accent); }
footer { text-align: center; color: var(--muted); font-size: 0.75rem; margin-top: 2rem; padding-top: 1rem; border-top: 1px solid var(--border); }
</style>
</head>
<body>
<div class="container">
<header>
<h1>${escapeHtml(manifest.title)}</h1>
${subtitle}
${metaHtml}
</header>
<main>
${sections}
</main>
<footer>Generated by Zopu at ${escapeHtml(manifest.generatedAt)}</footer>
</div>
</body>
</html>`;
};
@@ -0,0 +1,259 @@
import { env } from "@code/env/agent";
/* eslint-disable max-classes-per-file -- service client and its error type form one domain module. */
/**
* Service-authenticated Convex client for runtime → Convex callbacks.
*
* The agent's timeline/context tools call through this client to persist agent
* messages, summaries, artifacts, and timeline cards. It POSTs to the
* service-authenticated Convex Site HTTP actions exposed in
* `packages/backend/convex/agentHttp.ts`:
*
* POST /api/agents/events/message
* POST /api/agents/events/artifact
* POST /api/agents/storage/generate-upload-url
*
* These actions carry the FLUE_DB_TOKEN bearer so Convex authenticates the call
* as the intelligence runtime service (not a browser session) and enforce the
* agent→project→organization boundary server-side. The runtime never holds
* database credentials and never calls Convex mutations directly.
*/
/**
* Shared error type for non-OK responses from Convex service actions.
* The Convex httpActions return `{ error }` with a status of 400/401 on
* validation/auth failures.
*/
class ConvexServiceError extends Error {
readonly status: number;
constructor(message: string, status: number) {
super(message);
this.name = "ConvexServiceError";
this.status = status;
}
}
/**
* Boundary context carried in every runtime → Convex callback. The Convex
* `resolveBoundary` helper validates these against the conversationAgents row.
*/
interface BoundaryInput {
agentId: string;
organizationId: string;
projectId: string;
}
export class ConvexServiceClient {
private readonly baseUrl: string;
private readonly token: string;
constructor() {
this.baseUrl = env.CONVEX_SITE_URL.replace(/\/$/u, "");
this.token = env.FLUE_DB_TOKEN;
}
/**
* POST an agent-authored message to the global timeline.
* Maps to the Convex `events:appendToolEvent` mutation via the
* `/api/agents/events/message` Site action. Idempotent by idempotencyKey.
*/
async postAgentMessage(input: {
organizationId: string;
projectId: string;
agentId: string;
text: string;
idempotencyKey: string;
correlationId: string;
type: string;
visibility: string;
replyToEventId?: string;
workId?: string;
threadId?: string;
}): Promise<{ eventId: string }> {
const result = await this.postJson("/api/agents/events/message", {
agentId: input.agentId,
correlationId: input.correlationId,
idempotencyKey: input.idempotencyKey,
organizationId: input.organizationId,
projectId: input.projectId,
replyToEventId: input.replyToEventId,
text: input.text,
threadId: input.threadId,
type: input.type,
visibility: input.visibility,
workId: input.workId,
});
return { eventId: result.eventId as string };
}
/**
* Upload rendered HTML (or other bytes) to Convex Storage and return the
* storageId. The flow is:
* 1. POST /api/agents/storage/generate-upload-url → one-time upload URL.
* 2. PUT the bytes as a Blob to the upload URL.
* 3. The Convex-generated storageId is the UUID in the upload URL path.
*
* Returns undefined if the upload could not be completed so the caller can
* still publish an inline-only artifact.
*/
async uploadHtml(
boundary: BoundaryInput,
html: string
): Promise<{ storageId: string } | undefined> {
// 1. Obtain a one-time upload URL.
const urlResult = await this.postJson(
"/api/agents/storage/generate-upload-url",
{
agentId: boundary.agentId,
organizationId: boundary.organizationId,
projectId: boundary.projectId,
}
);
const uploadUrl = urlResult.url as string | undefined;
if (!uploadUrl) {
return undefined;
}
// 2. PUT the bytes as a Blob to the upload URL.
const blob = new Blob([html], { type: "text/html" });
const uploadResponse = await fetch(uploadUrl, {
body: blob,
method: "POST",
});
if (!uploadResponse.ok) {
return undefined;
}
// 3. The Convex-generated storageId is in the upload URL's path
// (/upload/<uuid>) and is echoed back in the JSON body of the PUT.
const storageId = await uploadResponse
.json()
.then((body) => (body as { storageId?: string }).storageId)
.catch(() => {
/* malformed upload response body — fall through to URL fallback */
});
if (storageId) {
return { storageId };
}
// Fallback: extract the UUID from the upload URL path.
const match = /\/upload\/(?<id>[A-Za-z0-9_-]+)/u.exec(uploadUrl);
return match?.groups?.id ? { storageId: match.groups.id } : undefined;
}
/**
* Publish a typed artifact revision (with an optional Convex Storage id for
* rendered HTML) and emit the `artifact.published` timeline event. Maps to
* the Convex `artifacts:publishRevision` mutation + `events:appendToolEvent`
* via the `/api/agents/events/artifact` Site action.
*
* Note: the artifact revision requires a `createdByEventId` (the timeline
* event that produced it). The onboarding slice publishes the message event
* first, then uses its eventId here.
*/
async publishArtifact(input: {
organizationId: string;
projectId: string;
agentId: string;
createdByEventId: string;
kind: string;
logicalKey: string;
title: string;
summary: string;
status: string;
card: unknown;
content: unknown;
storageId?: string;
idempotencyKey: string;
correlationId: string;
workId?: string;
threadId?: string;
}): Promise<{ eventId: string; artifactId: string }> {
const result = await this.postJson("/api/agents/events/artifact", {
agentId: input.agentId,
card: input.card,
content: input.content,
correlationId: input.correlationId,
createdByEventId: input.createdByEventId,
idempotencyKey: input.idempotencyKey,
kind: input.kind,
logicalKey: input.logicalKey,
organizationId: input.organizationId,
projectId: input.projectId,
status: input.status,
storageId: input.storageId,
summary: input.summary,
threadId: input.threadId,
title: input.title,
workId: input.workId,
});
return {
artifactId: result.artifactId as string,
eventId: result.eventId as string,
};
}
/**
* Store a project context document (e.g. the onboarding summary).
* Published as an inline artifact revision of kind "summary" so it persists
* durably and is recoverable across agent restarts.
*/
upsertProjectSummary(input: {
organizationId: string;
projectId: string;
agentId: string;
kind: string;
content: string;
correlationId: string;
createdByEventId: string;
idempotencyKey: string;
}): Promise<{ eventId: string; artifactId: string }> {
return this.publishArtifact({
agentId: input.agentId,
card: { kind: input.kind },
content: { markdown: input.content },
correlationId: input.correlationId,
createdByEventId: input.createdByEventId,
idempotencyKey: input.idempotencyKey,
kind: "summary",
logicalKey: `project:${input.projectId}:summary`,
organizationId: input.organizationId,
projectId: input.projectId,
status: "ready",
summary: input.content.slice(0, 160),
title: "Project summary",
});
}
/**
* POST JSON to a Convex Site action with the service bearer token.
* Returns the parsed JSON body. Throws ConvexServiceError on non-2xx.
*/
private async postJson(
path: string,
body: Record<string, unknown>
): Promise<Record<string, unknown>> {
const response = await fetch(`${this.baseUrl}${path}`, {
body: JSON.stringify(body),
headers: {
authorization: `Bearer ${this.token}`,
"content-type": "application/json",
},
method: "POST",
});
if (!response.ok) {
const detail = await response
.json()
.then((b) => (b as { error?: string }).error)
.catch(() => {
/* non-JSON error body — fall through to status fallback */
});
throw new ConvexServiceError(
detail ?? `Convex action ${path} returned ${response.status}`,
response.status
);
}
return (await response.json()) as Record<string, unknown>;
}
}
@@ -0,0 +1,77 @@
"use agent";
import { useModel, useTool } from "@flue/runtime";
import { modelSpecifier } from "../model-config.ts";
import { projectContextUpdate } from "../tools/project-context.ts";
import {
sandboxDescribe,
sandboxListFiles,
sandboxReadFile,
sandboxSearch,
} from "../tools/sandbox.ts";
import {
timelinePostMessage,
timelinePublishArtifact,
} from "../tools/timeline.ts";
/**
* ProjectConversationAgent — one project-bound Flue 2.0 conversation agent.
*
* Identity is stable and structural: `conversation:<organizationId>:<projectId>:v1`.
* The agent instance id passed to dispatch() encodes these facts:
* conversation:<organizationId>:<projectId>
*
* On a `project.ready` signal it:
* 1. Posts one idempotent short acknowledgement.
* 2. Explores the repository via VM read-only tools.
* 3. Stores a concise summary via projectContextUpdate.
* 4. Renders a static setup artifact and publishes it to the timeline.
*
* The agent is read-oriented for the first slice — no autonomous code mutation.
*/
export const ProjectConversationAgent = () => {
useModel(modelSpecifier());
// Timeline tools — constrained write surface to Convex (not arbitrary mutations).
useTool(timelinePostMessage);
useTool(timelinePublishArtifact);
// Project context tools — store durable project knowledge.
useTool(projectContextUpdate);
// Sandbox tools — read-only repository access through the project VM.
useTool(sandboxDescribe);
useTool(sandboxListFiles);
useTool(sandboxReadFile);
useTool(sandboxSearch);
return `You are Zopu's project conversation agent — an effective coworker who already understands this project.
You are bound to exactly one project and its repository workspace VM. You have read-only access to the repository through your tools.
## Onboarding (project.ready)
When you receive a \`project.ready\` signal:
1. Acknowledge briefly (1-2 sentences) using timelinePostMessage with idempotencyKey "dispatch:<dispatchId>:message:acknowledge".
2. Use sandboxDescribe to understand the repository structure.
3. Use sandboxListFiles on the root to see the layout.
4. Read key files: README.md, package.json, or equivalent entry documentation.
5. Synthesize a concise summary and store it via projectContextUpdate (kind "summary").
6. Render a polished setup artifact via timelinePublishArtifact with idempotencyKey "dispatch:<dispatchId>:message:onboarding-artifact".
## Style
- Default to 1-3 short sentences per response.
- Be specific and evidence-backed — cite file paths.
- Do not claim code was changed, tested, or shipped when it was not.
- For writing requests: inspect, scope, and explain — do not falsely claim completion.
- Keep internal exploration invisible; post only meaningful results to the timeline.
## Constraints
- You may NOT mutate repository source files. Output only goes under .zopu/.
- You must NOT expose API keys, credentials, or secrets.
- You must NOT access paths outside the repository root.`;
};
// Durable identity — decoupled from the source-level function name so
// minification or renames cannot corrupt conversation storage.
ProjectConversationAgent.agentName = "project-conversation";
+253
View File
@@ -0,0 +1,253 @@
import { env } from "@code/env/agent";
import { dispatch } from "@flue/runtime";
import { createAgentRouter } from "@flue/runtime/routing";
import { Hono } from "hono";
import {
cloneRepository,
getOrCreateProjectVm,
describeRepository,
scanEnvExample,
} from "./adapters/agentos.ts";
import { ProjectConversationAgent } from "./agents/project-agent.ts";
import { registerModelProvider } from "./model-config.ts";
import { registry } from "./runner.ts";
// Register the Cheaptricks OpenAI-compatible provider before the server
// starts handling requests. The API key is resolved per-request inside the
// provider's auth resolver and is never exposed to model tools.
registerModelProvider();
const app = new Hono();
// --- Health (unauthenticated liveness) --------------------------------------
app.get("/health", (c) => c.json({ service: "zopu-agents", status: "ok" }));
// --- Auth -------------------------------------------------------------------
/** Validate the bearer token against FLUE_DB_TOKEN. */
const requireServiceAuth = (authHeader: string | undefined): boolean => {
if (!authHeader) {
return false;
}
const match = /^Bearer\s+(?<token>.+)$/u.exec(authHeader);
if (!match) {
return false;
}
return match.groups?.token === env.FLUE_DB_TOKEN;
};
// --- Internal project-setup endpoint ----------------------------------------
//
// Called by the Convex `projectSetup.runSetup` coordinator to perform the
// AgentOS VM get-or-create, shallow clone, readability verification, and root
// `.env.example` name-only scan. Returns exactly the coordinator contract.
//
// Convex request body (callSetupEndpoint in projectSetup.ts):
// { branch, projectId, repositoryUrl }
//
// Coordinator-expected response (SetupRuntimeResult):
// { runtimeId?, vmId?, workspacePath?, repositoryPath?,
// repositoryCommit?, environmentVariableNames? }
interface ProjectSetupRequest {
branch: string;
projectId: string;
repositoryUrl: string;
}
export interface ProjectSetupResult {
runtimeId?: string;
vmId?: string;
workspacePath?: string;
repositoryPath?: string;
repositoryCommit?: string;
environmentVariableNames?: string[];
}
/**
* POST /internal/project-setup
*
* Single-call setup coordinator target. Performs:
* 1. Get-or-create the project workspace VM (idempotent by projectId).
* 2. Shallow single-branch clone (idempotent — skips if repo dir exists).
* 3. Verify readability: HEAD commit + root listing + read one file.
* 4. Scan root `.env.example` for variable names only (values never stored).
*
* Never mutates repository source files. The returned object matches the
* Convex `SetupRuntimeResult` wire shape exactly.
*/
app.post("/internal/project-setup", async (c) => {
if (!requireServiceAuth(c.req.header("authorization"))) {
return c.json({ error: "unauthorized" }, 401);
}
let body: ProjectSetupRequest;
try {
body = (await c.req.json()) as ProjectSetupRequest;
} catch {
return c.json({ error: "invalid JSON body" }, 400);
}
if (
typeof body.projectId !== "string" ||
body.projectId.length === 0 ||
typeof body.repositoryUrl !== "string" ||
body.repositoryUrl.length === 0 ||
typeof body.branch !== "string" ||
body.branch.length === 0
) {
return c.json({ error: "invalid project-setup request" }, 400);
}
try {
const vm = await getOrCreateProjectVm(body.projectId);
// Shallow clone (idempotent). Credentials are never exposed.
const { commit } = await cloneRepository({
branch: body.branch,
projectId: body.projectId,
repositoryUrl: body.repositoryUrl,
});
// Verify readability: HEAD commit + root listing (throws on failure).
await describeRepository(body.projectId);
// Scan root .env.example names-only.
const { names } = await scanEnvExample(body.projectId);
const result: ProjectSetupResult = {
environmentVariableNames: names,
repositoryCommit: commit,
repositoryPath: vm.repositoryPath,
runtimeId: vm.vmId,
vmId: vm.vmId,
workspacePath: vm.workspacePath,
};
return c.json(result, 200);
} catch (error) {
const message = error instanceof Error ? error.message : String(error);
return c.json({ detail: message, error: "project-setup failed" }, 500);
}
});
// --- Private dispatch endpoint -----------------------------------------------
//
// Convex calls this with a bearer FLUE_DB_TOKEN to deliver an event to a
// project-bound conversation agent. The endpoint validates the envelope,
// calls Flue dispatch(), and returns 202 Accepted.
//
// Idempotency: the dispatchId is encoded in the signal's idempotencyKey so
// a retried Convex workflow step converges on the original submission.
export interface DispatchEnvelope {
dispatchId: string;
// The triggering event. Convex sends sourceEventId separately for
// correlation; sourceEvent carries the event type + payload the agent acts on.
sourceEvent: {
type: string;
payload?: unknown;
};
// The Convex events row id of the source event (e.g. the project.ready id).
// Used for reply correlation; optional because some dispatches may be
// ad-hoc.
sourceEventId?: string;
agentId: string;
organizationId: string;
projectId: string;
// Correlation id tying all setup/dispatch events together.
correlationId?: string;
workId?: string;
threadId?: string;
}
app.post("/internal/agents/:agentId/events", async (c) => {
// Bearer auth.
if (!requireServiceAuth(c.req.header("authorization"))) {
return c.json({ error: "unauthorized" }, 401);
}
const routeAgentId = c.req.param("agentId");
let body: DispatchEnvelope;
try {
body = (await c.req.json()) as DispatchEnvelope;
} catch {
return c.json({ error: "invalid JSON body" }, 400);
}
// Validate the dispatch envelope.
if (
typeof body.dispatchId !== "string" ||
typeof body.agentId !== "string" ||
typeof body.organizationId !== "string" ||
typeof body.projectId !== "string" ||
!body.sourceEvent ||
typeof body.sourceEvent.type !== "string"
) {
return c.json({ error: "invalid dispatch envelope" }, 400);
}
// The route agentId must match the envelope agentId.
if (routeAgentId !== body.agentId) {
return c.json({ error: "agentId mismatch" }, 400);
}
// Build the stable conversation instance id. Must match
// conversationAgents.register: `conversation:<org>:<project>:v1`.
// The envelope agentId is already in that form.
const conversationId = body.agentId;
// Dispatch into Flue as a signal carrying the source event.
// The signal's attributes carry dispatch metadata so the agent's tools
// can resolve project/org/dispatch context during the turn.
try {
await dispatch(ProjectConversationAgent, {
id: conversationId,
idempotencyKey: `event:${body.dispatchId}:agent:${body.agentId}:handler:v1`,
message: {
attributes: {
agentId: body.agentId,
correlationId: body.correlationId ?? "",
dispatchId: body.dispatchId,
organizationId: body.organizationId,
projectId: body.projectId,
sourceEventId: body.sourceEventId ?? "",
threadId: body.threadId ?? "",
workId: body.workId ?? "",
},
body: JSON.stringify(body.sourceEvent),
kind: "signal",
type: body.sourceEvent.type,
},
});
return c.json(
{ accepted: true, conversationId, dispatchId: body.dispatchId },
202
);
} catch (error) {
const message = error instanceof Error ? error.message : String(error);
return c.json({ detail: message, error: "dispatch failed" }, 500);
}
});
// --- Public agent router (protected, optional) -------------------------------
// Mounted for direct conversation access. Protected by the same bearer auth.
// The dispatch-only path above is the primary entry for Convex → agent.
const agentSubApp = createAgentRouter(ProjectConversationAgent);
const protectedAgent = new Hono();
protectedAgent.use("*", async (c, next) => {
if (!requireServiceAuth(c.req.header("authorization"))) {
return c.json({ error: "unauthorized" }, 401);
}
return await next();
});
protectedAgent.route("/", agentSubApp as unknown as Hono);
app.route("/agents/project-conversation", protectedAgent);
// --- In-process RivetKit registry (serverless handler) -----------------------
//
// The AgentOS registry runs in serverless mode inside this same process.
// Mounted on a private path so only internal callers (and the Engine via
// configurePool.url) reach it. registry.handler drives the per-request
// serverless runtime; importing the registry object performs no start.
app.all("/internal/rivet/*", (c) => registry.handler(c.req.raw));
export default app;
+10
View File
@@ -0,0 +1,10 @@
import { sqlite } from "@flue/runtime/node";
// Flue 2.0 file-backed SQLite persistence adapter.
// Stores canonical conversation state and accepted submissions for the
// project-bound conversation agent. Application product truth remains Convex.
//
// Default path: /srv/zopu/data/flue/flue.db (overridable via FLUE_DB_PATH).
export default sqlite(
process.env.FLUE_DB_PATH ?? "/srv/zopu/data/flue/flue.db"
);
+16
View File
@@ -0,0 +1,16 @@
// @code/agents — Zopu Flue 2.0 + Hono + AgentOS intelligence runtime.
//
// The Flue 2.0 Node target generates dist/server.mjs via @flue/vite. That
// generated entry calls startFlueNodeServer(), which loads src/app.ts (the
// Hono route map) and src/db.ts (SQLite persistence) internally.
//
// This package's public exports are the reusable pieces: the conversation
// agent definition and the model configuration utilities. The server entry
// is dist/server.mjs (generated, not hand-authored).
export { ProjectConversationAgent } from "./agents/project-agent.ts";
export { registerModelProvider, modelSpecifier } from "./model-config.ts";
export type {
ArtifactManifest,
ArtifactManifestSection,
} from "./adapters/artifact-renderer.ts";
export type { DispatchEnvelope, ProjectSetupResult } from "./app.ts";
+58
View File
@@ -0,0 +1,58 @@
import { env } from "@code/env/agent";
import { createProvider } from "@earendil-works/pi-ai";
import type { Provider } from "@earendil-works/pi-ai";
import { openAICompletionsApi } from "@earendil-works/pi-ai/api/openai-completions.lazy";
import { setProvider } from "@flue/runtime";
/**
* Register the Cheaptricks OpenAI-compatible provider with the Flue runtime.
*
* The provider is keyed by env.AGENT_MODEL_PROVIDER (default "cheaptricks"),
* exposing the model named by env.AGENT_MODEL_NAME (default "mimo-v2.5") over
* the OpenAI Completions API against env.AGENT_MODEL_BASE_URL.
*
* The API key is resolved at request time from env.AGENT_MODEL_API_KEY and is
* never exposed to model-facing tools — it lives only in the provider's auth
* resolver, not in agent instructions, tool inputs, or conversation state.
*/
export const registerModelProvider = (): Provider<"openai-completions"> => {
const provider = createProvider({
api: openAICompletionsApi(),
auth: {
apiKey: {
name: "Cheaptricks API key",
resolve: () =>
Promise.resolve({
auth: {
apiKey: env.AGENT_MODEL_API_KEY,
baseUrl: env.AGENT_MODEL_BASE_URL,
},
}),
},
},
baseUrl: env.AGENT_MODEL_BASE_URL,
id: env.AGENT_MODEL_PROVIDER,
models: [
{
api: "openai-completions",
baseUrl: env.AGENT_MODEL_BASE_URL,
contextWindow: env.AGENT_MODEL_CONTEXT_WINDOW,
cost: { cacheRead: 0, cacheWrite: 0, input: 0, output: 0 },
id: env.AGENT_MODEL_NAME,
input: ["text"],
maxTokens: env.AGENT_MODEL_MAX_TOKENS,
name: env.AGENT_MODEL_NAME,
provider: env.AGENT_MODEL_PROVIDER,
reasoning: false,
},
],
name: "Cheaptricks",
});
setProvider(provider);
return provider;
};
/** Full model specifier string for useModel(): provider-id/model-id. */
export const modelSpecifier = (): string =>
`${env.AGENT_MODEL_PROVIDER}/${env.AGENT_MODEL_NAME}`;
+65
View File
@@ -0,0 +1,65 @@
import git from "@agentos-software/git";
import { env } from "@code/env/agent";
import { agentOS, setup } from "@rivet-dev/agentos";
/**
* AgentOS registry — owns the durable project workspace actor.
*
* The actor is keyed structurally by [projectId] (project ids are globally
* unique), so each project resolves to one dedicated VM across the process's
* lifetime. The registry runs in serverless runtime mode and is served
* in-process by the Hono app (see app.ts), which mounts
* `registry.handler(c.req.raw)` at `/internal/rivet/*`. There is no
* standalone runner process and no `startAndWait` entry point.
*
* `configurePool.url` is the distinct endpoint the Engine invokes to reach
* this registry handler; it is NOT the Engine control-plane URL (that stays
* `endpoint: env.RIVET_ENDPOINT`). The actor still authenticates every
* connection via `onBeforeConnect` against RIVET_WORKSPACE_TOKEN, so only
* this deployment's adapter can drive the actor.
*
* Software: the actor bundles Git (@agentos-software/git) alongside the
* AgentOS default utilities. defaultSoftware alone is common utilities, not
* Git — without the explicit git entry, clone/rev-parse/grep would fail.
*
* Import-safe: this module exports a pure registry object and performs no
* top-level process start. Importing it (including `import type`) does not
* start a registry.
*/
// Project workspace actor — one VM per project, with Git + default utilities.
const projectWorkspace = agentOS<undefined, { token: string }>({
defaultSoftware: true,
onBeforeConnect: (_context, params) => {
if (params.token !== env.RIVET_WORKSPACE_TOKEN) {
throw new Error("Unauthorized workspace connection");
}
},
// Git is required for clone, rev-parse, and read-only repository inspection.
// defaultSoftware provides sh/coreutils/common utilities only.
software: [git],
});
// Internal mount path shared by the Hono app route and the registry's
// serverless base path. Both MUST agree — the handler uses this base path to
// route actor/metadata requests, and the app must mount it at the same path.
export const RIVET_SERVERLESS_BASE_PATH = "/internal/rivet";
// Registry: the single actor type served by this process, in serverless mode.
// `serverless.basePath` matches the app route mount point; `configurePool.url`
// is the address the Engine calls back into this handler. Importing this
// module does not start the registry — the handler is driven per-request by
// the Hono app.
export const registry = setup({
configurePool: {
url: env.RIVET_SERVERLESS_ENDPOINT,
},
endpoint: env.RIVET_ENDPOINT,
logging: { level: "info" },
serverless: {
basePath: RIVET_SERVERLESS_BASE_PATH,
},
use: {
projectWorkspace,
},
});
@@ -0,0 +1,63 @@
import { defineTool, useDelivery } from "@flue/runtime";
import * as v from "valibot";
import { ConvexServiceClient } from "../adapters/convex-service-client.ts";
let client: ConvexServiceClient | null = null;
const getClient = (): ConvexServiceClient => {
if (!client) {
client = new ConvexServiceClient();
}
return client;
};
/**
* projectContextUpdate — store durable project knowledge (summary documents).
*
* Published as an inline artifact revision of kind "summary" via the Convex
* `/api/agents/events/artifact` Site action, so it persists durably in Convex
* and is recoverable across agent restarts. Requires a creation event (the
* current dispatch message) as the `createdByEventId`.
*/
export const projectContextUpdate = defineTool({
description:
"Store or update a project context document (e.g. the onboarding summary). The document persists in Convex and is recoverable across agent restarts.",
input: v.object({
content: v.pipe(
v.string(),
v.description("Markdown content of the document")
),
kind: v.pipe(v.string(), v.description("Document kind, e.g. 'summary'")),
}),
name: "projectContextUpdate",
run: async (ctx) => {
// eslint-disable-next-line react-hooks/rules-of-hooks -- useDelivery is a Flue runtime hook, not a React hook; called inside the tool's run callback per Flue tool contract.
const delivery = useDelivery();
if (delivery.kind !== "signal") {
throw new Error("projectContextUpdate requires a signal delivery");
}
const attrs = delivery.attributes ?? {};
const { organizationId, projectId, agentId, correlationId, sourceEventId } =
attrs;
if (
!organizationId ||
!projectId ||
!agentId ||
!correlationId ||
!sourceEventId
) {
throw new Error("Delivery context missing source event metadata");
}
await getClient().upsertProjectSummary({
agentId,
content: ctx.data.content,
correlationId,
createdByEventId: sourceEventId,
idempotencyKey: `project:${projectId}:summary:${ctx.data.kind}`,
kind: ctx.data.kind,
organizationId,
projectId,
});
return { output: { stored: true } };
},
});
+141
View File
@@ -0,0 +1,141 @@
import { defineTool, useDelivery } from "@flue/runtime";
import * as v from "valibot";
import {
describeRepository,
listFiles,
readFile,
searchRepository,
} from "../adapters/agentos.ts";
/**
* Sandbox tools — read-only repository access through the project VM.
* These are model-callable Flue tools that operate inside the agent's turn.
*
* Tool input schemas use Valibot (Flue 2.0's schema standard), not Typebox.
*
* The project context (projectId) is extracted from the delivery signal's
* attributes, which carry the dispatch metadata.
*/
/** Extract dispatch metadata from the current delivery signal. */
const getDispatchMeta = (): {
projectId: string;
organizationId: string;
agentId: string;
dispatchId: string;
correlationId: string;
} => {
// eslint-disable-next-line react-hooks/rules-of-hooks -- useDelivery is a Flue runtime hook, not a React hook; called inside a tool helper per Flue tool contract.
const delivery = useDelivery();
if (delivery.kind !== "signal") {
throw new Error("Sandbox tools require a signal delivery");
}
const attrs = delivery.attributes ?? {};
const { projectId } = attrs;
const { organizationId } = attrs;
const { agentId } = attrs;
const { dispatchId } = attrs;
const { correlationId } = attrs;
if (
!projectId ||
!organizationId ||
!agentId ||
!dispatchId ||
!correlationId
) {
throw new Error("No projectId in delivery context");
}
return { agentId, correlationId, dispatchId, organizationId, projectId };
};
/** sandboxDescribe: report workspace/repo paths and current HEAD commit. */
export const sandboxDescribe = defineTool({
description:
"Describe the project repository workspace: paths and current HEAD commit. Call this first when exploring a repository.",
input: v.object({}),
name: "sandboxDescribe",
run: async () => {
const { projectId } = getDispatchMeta();
const desc = await describeRepository(projectId);
const rootEntries = desc.rootEntries.map((e) => ({
kind: e.kind,
name: e.name,
size: e.size,
}));
return {
output: {
commit: desc.commit,
repositoryPath: "/workspace/repo",
rootEntries,
},
};
},
});
/** sandboxListFiles: list directory entries relative to repo root. */
export const sandboxListFiles = defineTool({
description:
"List files and directories at a path relative to the repository root.",
input: v.object({
path: v.pipe(
v.string(),
v.description(
"Directory path relative to repository root (use '' for root)"
)
),
}),
name: "sandboxListFiles",
run: async (ctx) => {
const { projectId } = getDispatchMeta();
const result = await listFiles(projectId, ctx.data.path);
const entries = result.map((e) => ({
kind: e.kind,
name: e.name,
size: e.size,
}));
return { output: { entries } };
},
});
/** sandboxReadFile: read a file's content relative to repo root. */
export const sandboxReadFile = defineTool({
description:
"Read the text content of a file at a path relative to the repository root.",
input: v.object({
path: v.pipe(
v.string(),
v.description("File path relative to repository root")
),
}),
name: "sandboxReadFile",
run: async (ctx) => {
const { projectId } = getDispatchMeta();
const result = await readFile(projectId, ctx.data.path);
// Truncate very large files to keep context bounded.
const content =
result.content.length > 50_000
? `${result.content.slice(0, 50_000)}\n...[truncated]`
: result.content;
return { output: { content } };
},
});
/** sandboxSearch: search repository for a text query. */
export const sandboxSearch = defineTool({
description:
"Search the repository for a text query. Returns matching file paths and snippets.",
input: v.object({
query: v.pipe(v.string(), v.description("Search query (text or pattern)")),
}),
name: "sandboxSearch",
run: async (ctx) => {
const { projectId } = getDispatchMeta();
const result = await searchRepository(projectId, ctx.data.query);
const matches = result.matches.map((m) => ({
path: m.path,
snippet: m.snippet,
}));
return { output: { matches } };
},
});
+185
View File
@@ -0,0 +1,185 @@
import { defineTool, useDelivery } from "@flue/runtime";
import * as v from "valibot";
import { renderStaticHtml } from "../adapters/artifact-renderer.ts";
import { ConvexServiceClient } from "../adapters/convex-service-client.ts";
let client: ConvexServiceClient | null = null;
const getClient = (): ConvexServiceClient => {
if (!client) {
client = new ConvexServiceClient();
}
return client;
};
/** Extract dispatch/project context from the current delivery signal. */
const getDispatchContext = (): {
dispatchId: string;
correlationId: string;
organizationId: string;
projectId: string;
agentId: string;
sourceEventId: string;
} => {
// eslint-disable-next-line react-hooks/rules-of-hooks -- useDelivery is a Flue runtime hook, not a React hook; called inside a tool helper per Flue tool contract.
const delivery = useDelivery();
if (delivery.kind !== "signal") {
throw new Error("Timeline tools require a signal delivery");
}
const attrs = delivery.attributes ?? {};
const { dispatchId } = attrs;
const { correlationId } = attrs;
const { organizationId } = attrs;
const { projectId } = attrs;
const { agentId } = attrs;
const { sourceEventId } = attrs;
if (
!dispatchId ||
!correlationId ||
!organizationId ||
!projectId ||
!agentId ||
!sourceEventId
) {
throw new Error("Delivery context missing source event metadata");
}
return {
agentId,
correlationId,
dispatchId,
organizationId,
projectId,
sourceEventId,
};
};
/**
* timelinePostMessage — post an agent-authored message to the global timeline.
* The message is idempotent by the provided idempotencyKey; a retry with the
* same key converges on the original event.
*
* Posts through the Convex `/api/agents/events/message` Site action, which
* enforces the agent→project→organization boundary and exact-once semantics.
*/
export const timelinePostMessage = defineTool({
description:
"Post a short message to the project timeline. Use for acknowledgements and brief responses (1-3 sentences). Provide a unique idempotencyKey.",
input: v.object({
idempotencyKey: v.pipe(
v.string(),
v.description("Stable idempotency key for this message")
),
importance: v.optional(
v.pipe(
v.union([v.literal("normal"), v.literal("high")]),
v.description("Message importance level (default: normal)")
)
),
text: v.pipe(
v.string(),
v.description("The message text to post (1-3 sentences)")
),
}),
name: "timelinePostMessage",
run: async (ctx) => {
const { correlationId, organizationId, projectId, agentId, sourceEventId } =
getDispatchContext();
const result = await getClient().postAgentMessage({
agentId,
correlationId,
idempotencyKey: ctx.data.idempotencyKey,
organizationId,
projectId,
replyToEventId: sourceEventId,
text: ctx.data.text,
type: "agent.message",
visibility: "timeline",
});
return { output: { eventId: result.eventId } };
},
});
/**
* timelinePublishArtifact — render and publish a static setup artifact.
*
* Two-step flow matching the Convex artifact contract:
* 1. Post an acknowledgement/creation message (the `createdByEventId`) so the
* artifact revision is tied to a timeline event.
* 2. Render safe static HTML, upload it to Convex Storage, and publish the
* artifact revision referencing the uploaded storageId.
*
* The artifact is idempotent by logicalKey + version (server-side) and the
* event is exact-once by idempotencyKey.
*/
export const timelinePublishArtifact = defineTool({
description:
"Render a project setup artifact from structured sections and publish it to the timeline. Provide title, subtitle, and content sections.",
input: v.object({
idempotencyKey: v.pipe(
v.string(),
v.description("Stable idempotency key for this artifact")
),
sections: v.array(
v.object({
body: v.string(),
items: v.optional(v.array(v.string())),
title: v.string(),
})
),
subtitle: v.optional(
v.pipe(v.string(), v.description("Artifact subtitle"))
),
title: v.pipe(v.string(), v.description("Artifact title")),
}),
name: "timelinePublishArtifact",
run: async (ctx) => {
const { correlationId, organizationId, projectId, agentId } =
getDispatchContext();
// Step 1: create the timeline event that the artifact revision is tied to.
const createdBy = await getClient().postAgentMessage({
agentId,
correlationId,
idempotencyKey: `${ctx.data.idempotencyKey}:create`,
organizationId,
projectId,
text: `Published artifact: ${ctx.data.title}`,
type: "artifact.created",
visibility: "compact",
});
// Step 2: render + upload + publish the artifact revision.
const manifest = {
generatedAt: new Date().toISOString(),
kind: "project_setup",
sections: ctx.data.sections,
subtitle: ctx.data.subtitle,
title: ctx.data.title,
};
const html = renderStaticHtml(manifest);
const boundary = { agentId, organizationId, projectId };
const uploaded = await getClient().uploadHtml(boundary, html);
const result = await getClient().publishArtifact({
agentId,
card: {
sections: ctx.data.sections,
subtitle: ctx.data.subtitle,
},
content: { manifest },
correlationId,
createdByEventId: createdBy.eventId,
idempotencyKey: ctx.data.idempotencyKey,
kind: "project_setup",
logicalKey: `project:${projectId}:setup`,
organizationId,
projectId,
status: "ready",
storageId: uploaded?.storageId,
summary: ctx.data.subtitle ?? ctx.data.title,
title: ctx.data.title,
});
return {
output: { artifactId: result.artifactId, eventId: result.eventId },
};
},
});
+12
View File
@@ -0,0 +1,12 @@
{
"extends": "@code/config/tsconfig.base.json",
"compilerOptions": {
"outDir": "dist",
"rootDir": "src",
"allowImportingTsExtensions": true,
"noEmit": true,
"types": ["node"]
},
"include": ["src/**/*.ts", "*.d.ts"],
"exclude": ["node_modules", "dist"]
}
+22
View File
@@ -0,0 +1,22 @@
import { flue } from "@flue/vite";
import { defineConfig } from "vite";
// Flue 2.0 Node target. @flue/vite auto-discovers flue.config.ts, scans
// 'use agent' modules, and builds dist/server.mjs. The built server listens
// on 127.0.0.1:3000 (Caddy terminates TLS upstream at agents.zopu.puter.wtf).
export default defineConfig({
plugins: [
...flue({
providers: [],
target: "node",
}),
],
preview: {
host: process.env.HOST ?? "127.0.0.1",
port: Number(process.env.PORT ?? 3000),
},
server: {
host: process.env.HOST ?? "127.0.0.1",
port: Number(process.env.PORT ?? 3000),
},
});
+26
View File
@@ -8,10 +8,17 @@
* @module
*/
import type * as agentDispatches from "../agentDispatches.js";
import type * as agentHttp from "../agentHttp.js";
import type * as artifactBuilds from "../artifactBuilds.js";
import type * as artifacts from "../artifacts.js";
import type * as auth from "../auth.js";
import type * as authz from "../authz.js";
import type * as cleanup from "../cleanup.js";
import type * as conversationAgents from "../conversationAgents.js";
import type * as crons from "../crons.js";
import type * as events from "../events.js";
import type * as flowRuns from "../flowRuns.js";
import type * as gitConnectionData from "../gitConnectionData.js";
import type * as gitConnectionHealth from "../gitConnectionHealth.js";
import type * as gitConnections from "../gitConnections.js";
@@ -21,8 +28,14 @@ import type * as healthCheck from "../healthCheck.js";
import type * as http from "../http.js";
import type * as organizations from "../organizations.js";
import type * as privateData from "../privateData.js";
import type * as projectEnvironmentVariables from "../projectEnvironmentVariables.js";
import type * as projectRuntimes from "../projectRuntimes.js";
import type * as projectSetup from "../projectSetup.js";
import type * as projectSetupQueries from "../projectSetupQueries.js";
import type * as projects from "../projects.js";
import type * as publicGit from "../publicGit.js";
import type * as timeline from "../timeline.js";
import type * as works from "../works.js";
import type {
ApiFromModules,
@@ -31,10 +44,17 @@ import type {
} from "convex/server";
declare const fullApi: ApiFromModules<{
agentDispatches: typeof agentDispatches;
agentHttp: typeof agentHttp;
artifactBuilds: typeof artifactBuilds;
artifacts: typeof artifacts;
auth: typeof auth;
authz: typeof authz;
cleanup: typeof cleanup;
conversationAgents: typeof conversationAgents;
crons: typeof crons;
events: typeof events;
flowRuns: typeof flowRuns;
gitConnectionData: typeof gitConnectionData;
gitConnectionHealth: typeof gitConnectionHealth;
gitConnections: typeof gitConnections;
@@ -44,8 +64,14 @@ declare const fullApi: ApiFromModules<{
http: typeof http;
organizations: typeof organizations;
privateData: typeof privateData;
projectEnvironmentVariables: typeof projectEnvironmentVariables;
projectRuntimes: typeof projectRuntimes;
projectSetup: typeof projectSetup;
projectSetupQueries: typeof projectSetupQueries;
projects: typeof projects;
publicGit: typeof publicGit;
timeline: typeof timeline;
works: typeof works;
}>;
/**
+3
View File
@@ -25,6 +25,9 @@ import type { DataModel } from "./dataModel.js";
* Typesafe environment variables declared in `convex.config.ts`.
*/
type Env = {
readonly AGENT_BACKEND_URL: string | undefined;
readonly FLUE_DB_TOKEN: string | undefined;
readonly FLUE_URL: string | undefined;
readonly GITEA_TOKEN: string | undefined;
readonly GITEA_URL: string | undefined;
readonly GITEA_WEBHOOK_SECRET: string | undefined;
+184
View File
@@ -0,0 +1,184 @@
import { ConvexError, v } from "convex/values";
import type { Doc } from "./_generated/dataModel";
import { internalMutation, internalQuery } from "./_generated/server";
const dispatchStatus = v.union(
v.literal("pending"),
v.literal("sending"),
v.literal("accepted"),
v.literal("completed"),
v.literal("failed")
);
/**
* The handler-version stamp every dispatch carries. Bumped only when the
* handler contract changes; encoded into the idempotency key so a new version
* is a fresh dispatch rather than a duplicate of the old one.
*/
export const HANDLER_VERSION = "v1";
/**
* Create an AgentDispatch row keyed idempotently by
* `event:<sourceEventId>:agent:<agentId>:handler:<version>`. Retries reuse the
* existing pending/sending dispatch rather than duplicating deliveries.
* Returns `{ dispatchId, created }` so callers can decide whether to send.
*/
export const create = internalMutation({
args: {
agentId: v.string(),
idempotencyKey: v.string(),
organizationId: v.id("organizations"),
projectId: v.optional(v.id("projects")),
sourceEventId: v.id("events"),
threadId: v.optional(v.id("threads")),
workId: v.optional(v.id("works")),
},
handler: async (
ctx,
args
): Promise<{ created: boolean; dispatch: Doc<"agentDispatches"> }> => {
const existing = await ctx.db
.query("agentDispatches")
.withIndex("by_organizationId_and_source_event_handler", (q) =>
q
.eq("organizationId", args.organizationId)
.eq("sourceEventId", args.sourceEventId)
.eq("agentId", args.agentId)
.eq("handlerVersion", HANDLER_VERSION)
)
.unique();
if (existing) {
return { created: false, dispatch: existing };
}
const timestamp = Date.now();
const dispatchId = await ctx.db.insert("agentDispatches", {
agentId: args.agentId,
attempt: 1,
createdAt: timestamp,
handlerVersion: HANDLER_VERSION,
idempotencyKey: args.idempotencyKey,
organizationId: args.organizationId,
projectId: args.projectId,
sourceEventId: args.sourceEventId,
status: "pending",
threadId: args.threadId,
updatedAt: timestamp,
workId: args.workId,
});
const dispatch = await ctx.db.get(dispatchId);
if (!dispatch) {
throw new ConvexError("Agent dispatch could not be read after insert");
}
return { created: true, dispatch };
},
});
/**
* Transition a dispatch toward sending. Idempotent: a dispatch already accepted
* or completed is not rewound back to sending on retry.
*/
export const markSending = internalMutation({
args: { dispatchId: v.id("agentDispatches") },
handler: async (ctx, args): Promise<Doc<"agentDispatches">> => {
const dispatch = await ctx.db.get(args.dispatchId);
if (!dispatch) {
throw new ConvexError("Agent dispatch not found");
}
if (dispatch.status === "accepted" || dispatch.status === "completed") {
return dispatch;
}
await ctx.db.patch(args.dispatchId, {
status: "sending",
updatedAt: Date.now(),
});
const updated = await ctx.db.get(args.dispatchId);
if (!updated) {
throw new ConvexError("Agent dispatch could not be read after update");
}
return updated;
},
});
export const markAccepted = internalMutation({
args: { dispatchId: v.id("agentDispatches") },
handler: async (ctx, args): Promise<Doc<"agentDispatches">> => {
const dispatch = await ctx.db.get(args.dispatchId);
if (!dispatch) {
throw new ConvexError("Agent dispatch not found");
}
// Completed is terminal; do not rewind to accepted.
if (dispatch.status === "completed") {
return dispatch;
}
await ctx.db.patch(args.dispatchId, {
status: "accepted",
updatedAt: Date.now(),
});
const updated = await ctx.db.get(args.dispatchId);
if (!updated) {
throw new ConvexError("Agent dispatch could not be read after update");
}
return updated;
},
});
export const markCompleted = internalMutation({
args: { dispatchId: v.id("agentDispatches") },
handler: async (ctx, args): Promise<Doc<"agentDispatches">> => {
const dispatch = await ctx.db.get(args.dispatchId);
if (!dispatch) {
throw new ConvexError("Agent dispatch not found");
}
if (dispatch.status === "completed") {
return dispatch;
}
await ctx.db.patch(args.dispatchId, {
status: "completed",
updatedAt: Date.now(),
});
const updated = await ctx.db.get(args.dispatchId);
if (!updated) {
throw new ConvexError("Agent dispatch could not be read after update");
}
return updated;
},
});
export const markFailed = internalMutation({
args: {
dispatchId: v.id("agentDispatches"),
error: v.string(),
retry: v.boolean(),
},
handler: async (ctx, args): Promise<Doc<"agentDispatches">> => {
const dispatch = await ctx.db.get(args.dispatchId);
if (!dispatch) {
throw new ConvexError("Agent dispatch not found");
}
// Completed is terminal; a late failure after completion is ignored.
if (dispatch.status === "completed") {
return dispatch;
}
const nextAttempt = args.retry ? dispatch.attempt + 1 : dispatch.attempt;
await ctx.db.patch(args.dispatchId, {
attempt: nextAttempt,
lastError: args.error,
status: args.retry ? "pending" : "failed",
updatedAt: Date.now(),
});
const updated = await ctx.db.get(args.dispatchId);
if (!updated) {
throw new ConvexError("Agent dispatch could not be read after update");
}
return updated;
},
});
export const get = internalQuery({
args: { dispatchId: v.id("agentDispatches") },
handler: async (ctx, args): Promise<Doc<"agentDispatches"> | null> =>
await ctx.db.get(args.dispatchId),
});
export { dispatchStatus };
+291
View File
@@ -0,0 +1,291 @@
import { env } from "@code/env/convex";
import { makeFunctionReference } from "convex/server";
import { ConvexError } from "convex/values";
import type { Id } from "./_generated/dataModel";
import { httpAction } from "./_generated/server";
import type { ActionCtx } from "./_generated/server";
// never call Convex mutations directly; they POST to these HTTP endpoints and
// we run the validated mutations server-side.
const appendToolEventRef = makeFunctionReference<
"mutation",
{
actorTool: string;
artifactIds?: Id<"artifacts">[];
correlationId: string;
idempotencyKey: string;
organizationId: Id<"organizations">;
payload: unknown;
projectId?: Id<"projects">;
replyToEventId?: Id<"events">;
scopeKind: "global" | "work";
threadId?: Id<"threads">;
type: string;
visibility: "timeline" | "compact" | "internal";
workId?: Id<"works">;
},
Id<"events">
>("events:appendToolEvent");
const publishRevisionRef = makeFunctionReference<
"mutation",
{
card: unknown;
content: unknown;
createdByEventId: Id<"events">;
kind: string;
logicalKey: string;
organizationId: Id<"organizations">;
projectId?: Id<"projects">;
status: string;
storageId?: Id<"_storage">;
summary: string;
threadId?: Id<"threads">;
title: string;
workId?: Id<"works">;
},
{ _id: Id<"artifacts">; _creationTime: number }
>("artifacts:publishRevision");
const getAgentForProjectRef = makeFunctionReference<
"query",
{ projectId: Id<"projects"> },
{
_id: Id<"conversationAgents">;
id: string;
organizationId: Id<"organizations">;
projectId: Id<"projects">;
} | null
>("conversationAgents:getForProject");
/**
* Validate the `Authorization: Bearer <FLUE_DB_TOKEN>` header. Returns the
* token on success; throws a ConvexError (mapped to 401 by the httpAction).
*/
const requireServiceToken = (request: Request): string => {
if (!env.FLUE_DB_TOKEN) {
throw new ConvexError("FLUE_DB_TOKEN is not configured");
}
const header = request.headers.get("authorization") ?? "";
if (!header.startsWith("Bearer ")) {
throw new ConvexError("Missing bearer token");
}
const token = header.slice("Bearer ".length).trim();
if (token.length === 0 || token !== env.FLUE_DB_TOKEN) {
throw new ConvexError("Invalid service token");
}
return token;
};
interface BoundaryContext {
agentId: string;
organizationId: Id<"organizations">;
projectId: Id<"projects">;
}
/** Shape returned by the conversationAgents:getForProject internal query. */
interface AgentRow {
_id: Id<"conversationAgents">;
id: string;
organizationId: Id<"organizations">;
projectId: Id<"projects">;
}
/**
* Enforce the agent -> project -> organization boundary. The caller asserts an
* `agentId` of the form `conversation:<org>:<project>:v1`; we verify the
* conversationAgents row exists for that project and belongs to the claimed
* organization. This prevents a project-A agent from writing into project-B.
*/
const resolveBoundary = async (
ctx: ActionCtx,
body: { agentId?: unknown; organizationId?: unknown; projectId?: unknown }
): Promise<BoundaryContext> => {
if (typeof body.agentId !== "string" || body.agentId.length === 0) {
throw new ConvexError("agentId is required");
}
if (typeof body.projectId !== "string") {
throw new ConvexError("projectId is required");
}
if (typeof body.organizationId !== "string") {
throw new ConvexError("organizationId is required");
}
const projectId = body.projectId as Id<"projects">;
const organizationId = body.organizationId as Id<"organizations">;
const agent = (await ctx.runQuery(getAgentForProjectRef, {
projectId,
})) as AgentRow | null;
if (
!agent ||
agent.id !== body.agentId ||
agent.organizationId !== organizationId
) {
throw new ConvexError("Agent is not bound to this project/organization");
}
return { agentId: body.agentId, organizationId, projectId };
};
const json = (status: number, payload: unknown): Response =>
Response.json(payload, { status });
const asString = (value: unknown, field: string): string => {
if (typeof value !== "string" || value.length === 0) {
throw new ConvexError(`${field} is required`);
}
return value;
};
const asOptionalId = (value: unknown): string | undefined => {
if (typeof value !== "string" || value.length === 0) {
return undefined;
}
return value;
};
/**
* POST /api/agents/events/message
*
* Service-authenticated endpoint for a project conversation agent to append a
* timeline event (an assistant message, acknowledgement, etc.). Enforces the
* agent->project->organization boundary and exact-once semantics via the
* supplied idempotency key.
*
* Body: {
* agentId, organizationId, projectId,
* text, type, visibility, idempotencyKey, correlationId,
* replyToEventId?, workId?, threadId?
* }
*/
export const postAgentMessage = httpAction(async (ctx, request) => {
try {
requireServiceToken(request);
const body = (await request.json()) as Record<string, unknown>;
const boundary = await resolveBoundary(ctx, body);
const text = asString(body.text, "text");
const type = asString(body.type, "type");
const visibility = asString(body.visibility, "visibility");
const idempotencyKey = asString(body.idempotencyKey, "idempotencyKey");
const correlationId = asString(body.correlationId, "correlationId");
if (!["timeline", "compact", "internal"].includes(visibility)) {
return json(400, { error: "Invalid visibility" });
}
const eventId = (await ctx.runMutation(appendToolEventRef, {
actorTool: boundary.agentId,
artifactIds: undefined,
correlationId,
idempotencyKey,
organizationId: boundary.organizationId,
payload: { text },
projectId: boundary.projectId,
replyToEventId: asOptionalId(body.replyToEventId) as
| Id<"events">
| undefined,
scopeKind: "global",
threadId: asOptionalId(body.threadId) as Id<"threads"> | undefined,
type,
visibility: visibility as "timeline" | "compact" | "internal",
workId: asOptionalId(body.workId) as Id<"works"> | undefined,
})) as Id<"events">;
return json(200, { eventId });
} catch (error) {
const message = error instanceof Error ? error.message : String(error);
const status = message.startsWith("Unauthorized") ? 401 : 400;
return json(status, { error: message });
}
});
/**
* POST /api/agents/events/artifact
*
* Service-authenticated endpoint for a project conversation agent to publish a
* typed artifact revision (with an optional Convex Storage id) and emit the
* `artifact.published` event referencing it. Enforces the boundary and is
* idempotent through the artifact's logicalKey+version and the event key.
*
* Body: {
* agentId, organizationId, projectId,
* createdByEventId, kind, logicalKey, title, summary, status,
* card, content, storageId?, idempotencyKey, correlationId,
* workId?, threadId?
* }
*/
export const publishAgentArtifact = httpAction(async (ctx, request) => {
try {
requireServiceToken(request);
const body = (await request.json()) as Record<string, unknown>;
const boundary = await resolveBoundary(ctx, body);
const createdByEventId = asString(
body.createdByEventId,
"createdByEventId"
) as Id<"events">;
const kind = asString(body.kind, "kind");
const logicalKey = asString(body.logicalKey, "logicalKey");
const title = asString(body.title, "title");
const summary = asString(body.summary, "summary");
const status = asString(body.status, "status");
const idempotencyKey = asString(body.idempotencyKey, "idempotencyKey");
const correlationId = asString(body.correlationId, "correlationId");
const card = body.card ?? {};
const content = body.content ?? {};
const artifact = (await ctx.runMutation(publishRevisionRef, {
card,
content,
createdByEventId,
kind,
logicalKey,
organizationId: boundary.organizationId,
projectId: boundary.projectId,
status,
storageId: asOptionalId(body.storageId) as Id<"_storage"> | undefined,
summary,
threadId: asOptionalId(body.threadId) as Id<"threads"> | undefined,
title,
workId: asOptionalId(body.workId) as Id<"works"> | undefined,
})) as { _id: Id<"artifacts"> };
// Emit the artifact.published timeline event (exact-once).
const eventId = (await ctx.runMutation(appendToolEventRef, {
actorTool: boundary.agentId,
artifactIds: [artifact._id],
correlationId,
idempotencyKey: `${idempotencyKey}:event`,
organizationId: boundary.organizationId,
payload: { artifactId: artifact._id, title },
projectId: boundary.projectId,
scopeKind: "global",
threadId: asOptionalId(body.threadId) as Id<"threads"> | undefined,
type: "artifact.published",
visibility: "timeline",
workId: asOptionalId(body.workId) as Id<"works"> | undefined,
})) as Id<"events">;
return json(200, { artifactId: artifact._id, eventId });
} catch (error) {
const message = error instanceof Error ? error.message : String(error);
const status = message.startsWith("Unauthorized") ? 401 : 400;
return json(status, { error: message });
}
});
/**
* POST /api/agents/storage/generate-upload-url
*
* Service-authenticated endpoint for an agent to obtain a one-time Convex
* Storage upload URL for a binary artifact (e.g. rendered HTML). The agent
* uploads the bytes to the returned URL and then references the resulting
* storageId in {@link publishAgentArtifact}.
*
* Body: { agentId, organizationId, projectId }
*/
export const generateAgentUploadUrl = httpAction(async (ctx, request) => {
try {
requireServiceToken(request);
const body = (await request.json()) as Record<string, unknown>;
await resolveBoundary(ctx, body);
const url = await ctx.storage.generateUploadUrl();
return json(200, { url });
} catch (error) {
const message = error instanceof Error ? error.message : String(error);
const status = message.startsWith("Unauthorized") ? 401 : 400;
return json(status, { error: message });
}
});
+141
View File
@@ -0,0 +1,141 @@
import { ConvexError, v } from "convex/values";
import type { Doc } from "./_generated/dataModel";
import { internalMutation, internalQuery } from "./_generated/server";
const artifactKind = v.union(
v.literal("project_setup"),
v.literal("summary"),
v.literal("plan"),
v.literal("preview"),
v.literal("blocker"),
v.literal("report")
);
const buildStatus = v.union(
v.literal("pending"),
v.literal("building"),
v.literal("ready"),
v.literal("failed")
);
/**
* Create an ArtifactBuild keyed idempotently by
* `flow:<flowRunId>:artifact:<kind>:revision:<revision>` (per contract). The
* build records the typed source manifest; the rendered artifact + storage
* upload are linked via {@link markReady}.
*/
export const create = internalMutation({
args: {
artifactKind,
flowRunId: v.optional(v.id("flowRuns")),
idempotencyKey: v.string(),
organizationId: v.id("organizations"),
projectId: v.optional(v.id("projects")),
sourceManifest: v.any(),
threadId: v.optional(v.id("threads")),
workId: v.optional(v.id("works")),
},
handler: async (ctx, args): Promise<Doc<"artifactBuilds">> => {
// Idempotency: reuse an existing build for the same flow run + kind.
if (args.flowRunId) {
const existing = await ctx.db
.query("artifactBuilds")
.withIndex("by_organizationId_and_flow_run", (q) =>
q
.eq("organizationId", args.organizationId)
.eq("flowRunId", args.flowRunId)
)
.filter((q) => q.eq(q.field("artifactKind"), args.artifactKind))
.unique();
if (existing) {
return existing;
}
}
const timestamp = Date.now();
const buildId = await ctx.db.insert("artifactBuilds", {
artifactKind: args.artifactKind,
createdAt: timestamp,
flowRunId: args.flowRunId,
idempotencyKey: args.idempotencyKey,
organizationId: args.organizationId,
projectId: args.projectId,
sourceManifest: args.sourceManifest,
status: "pending",
threadId: args.threadId,
updatedAt: timestamp,
workId: args.workId,
});
const build = await ctx.db.get(buildId);
if (!build) {
throw new ConvexError("Artifact build could not be read after insert");
}
return build;
},
});
/**
* Link a finished artifact build to its rendered Convex Storage id and the
* created artifact row. A ready build is terminal.
*/
export const markReady = internalMutation({
args: {
artifactId: v.id("artifacts"),
buildId: v.id("artifactBuilds"),
outputStorageId: v.optional(v.id("_storage")),
},
handler: async (ctx, args): Promise<Doc<"artifactBuilds">> => {
const build = await ctx.db.get(args.buildId);
if (!build) {
throw new ConvexError("Artifact build not found");
}
if (build.status === "ready") {
return build;
}
await ctx.db.patch(args.buildId, {
artifactId: args.artifactId,
outputStorageId: args.outputStorageId,
status: "ready",
updatedAt: Date.now(),
});
const updated = await ctx.db.get(args.buildId);
if (!updated) {
throw new ConvexError("Artifact build could not be read after update");
}
return updated;
},
});
export const markFailed = internalMutation({
args: {
buildId: v.id("artifactBuilds"),
error: v.string(),
},
handler: async (ctx, args): Promise<Doc<"artifactBuilds">> => {
const build = await ctx.db.get(args.buildId);
if (!build) {
throw new ConvexError("Artifact build not found");
}
if (build.status === "ready") {
return build;
}
await ctx.db.patch(args.buildId, {
lastError: args.error,
status: "failed",
updatedAt: Date.now(),
});
const updated = await ctx.db.get(args.buildId);
if (!updated) {
throw new ConvexError("Artifact build could not be read after update");
}
return updated;
},
});
export const get = internalQuery({
args: { buildId: v.id("artifactBuilds") },
handler: async (ctx, args): Promise<Doc<"artifactBuilds"> | null> =>
await ctx.db.get(args.buildId),
});
export { buildStatus };
+89 -2
View File
@@ -1,7 +1,7 @@
import { v } from "convex/values";
import { ConvexError, v } from "convex/values";
import type { Doc } from "./_generated/dataModel";
import { query } from "./_generated/server";
import { internalMutation, query } from "./_generated/server";
import { requireArtifactMember, requireWorkMember } from "./authz";
export const get = query({
@@ -25,3 +25,90 @@ export const listForWork = query({
.take(limit);
},
});
/**
* Publish (or supersede) a typed artifact revision for a project/work/thread.
*
* Versioning: the new artifact's `version` is one greater than the latest
* existing artifact sharing the same `logicalKey` within the org. The previous
* latest artifact (if any) is marked `superseded` and linked via
* `supersedesArtifactId` on the new row.
*
* The optional `storageId` links the rendered static HTML (or other binary) in
* Convex Storage to the artifact row, keeping the card/manifest inline for the
* timeline.
*
* Idempotent by `logicalKey` + `version`: a re-publish of the same version
* returns the existing artifact. `createdByEventId` ties the artifact to the
* event that produced it (exact-once via the event's idempotency key).
*/
export const publishRevision = internalMutation({
args: {
card: v.any(),
content: v.any(),
createdByEventId: v.id("events"),
kind: v.union(
v.literal("project_setup"),
v.literal("summary"),
v.literal("plan"),
v.literal("preview"),
v.literal("blocker"),
v.literal("report")
),
logicalKey: v.string(),
organizationId: v.id("organizations"),
projectId: v.optional(v.id("projects")),
status: v.union(
v.literal("working"),
v.literal("ready"),
v.literal("blocked"),
v.literal("failed"),
v.literal("superseded")
),
storageId: v.optional(v.id("_storage")),
summary: v.string(),
threadId: v.optional(v.id("threads")),
title: v.string(),
workId: v.optional(v.id("works")),
},
handler: async (ctx, args): Promise<Doc<"artifacts">> => {
const previous = await ctx.db
.query("artifacts")
.withIndex("by_organizationId_and_logicalKey_and_version", (q) =>
q
.eq("organizationId", args.organizationId)
.eq("logicalKey", args.logicalKey)
)
.order("desc")
.first();
const version = (previous?.version ?? 0) + 1;
const now = Date.now();
const artifactId = await ctx.db.insert("artifacts", {
card: args.card,
content: args.content,
createdAt: now,
createdByEventId: args.createdByEventId,
kind: args.kind,
logicalKey: args.logicalKey,
organizationId: args.organizationId,
projectId: args.projectId,
status: args.status,
storageId: args.storageId,
summary: args.summary,
supersedesArtifactId: previous?._id,
threadId: args.threadId,
title: args.title,
version,
workId: args.workId,
});
// Mark the prior revision superseded so only the latest is "ready".
if (previous && previous.status !== "superseded") {
await ctx.db.patch(previous._id, { status: "superseded" });
}
const artifact = await ctx.db.get(artifactId);
if (!artifact) {
throw new ConvexError("Artifact could not be read after publish");
}
return artifact;
},
});
@@ -0,0 +1,104 @@
import { ConvexError, v } from "convex/values";
import type { Doc, Id } from "./_generated/dataModel";
import { internalMutation, internalQuery } from "./_generated/server";
const agentStatus = v.union(
v.literal("registered"),
v.literal("active"),
v.literal("idle"),
v.literal("disabled")
);
/**
* The stable conversation id used as the Flue `dispatch` conversation id and
* the `conversationAgents.id` uniqueness key.
* Shape: `conversation:<organizationId>:<projectId>:v1`
*/
export const conversationAgentId = (
organizationId: Id<"organizations">,
projectId: Id<"projects">
): string => `conversation:${organizationId}:${projectId}:v1`;
/**
* Register one project-bound conversation agent per Project. Idempotent by the
* stable `conversation:<org>:<project>:v1` key; retries reuse the existing row.
*/
export const register = internalMutation({
args: {
agentType: v.string(),
organizationId: v.id("organizations"),
projectId: v.id("projects"),
runtimeId: v.optional(v.id("projectRuntimes")),
},
handler: async (ctx, args): Promise<Doc<"conversationAgents">> => {
const id = conversationAgentId(args.organizationId, args.projectId);
const existing = await ctx.db
.query("conversationAgents")
.withIndex("by_projectId", (q) => q.eq("projectId", args.projectId))
.unique();
if (existing) {
return existing;
}
const timestamp = Date.now();
const rowId = await ctx.db.insert("conversationAgents", {
agentType: args.agentType,
createdAt: timestamp,
id,
organizationId: args.organizationId,
projectId: args.projectId,
runtimeId: args.runtimeId,
status: "registered",
updatedAt: timestamp,
});
const saved = await ctx.db.get(rowId);
if (!saved) {
throw new ConvexError(
"Conversation agent could not be read after insert"
);
}
return saved;
},
});
export const markStatus = internalMutation({
args: {
agentRowId: v.id("conversationAgents"),
flueConversationId: v.optional(v.string()),
lastEventId: v.optional(v.id("events")),
status: agentStatus,
},
handler: async (ctx, args): Promise<Doc<"conversationAgents">> => {
const agent = await ctx.db.get(args.agentRowId);
if (!agent) {
throw new ConvexError("Conversation agent not found");
}
const patch: Record<string, unknown> = {
status: args.status,
updatedAt: Date.now(),
};
if (args.flueConversationId !== undefined) {
patch.flueConversationId = args.flueConversationId;
}
if (args.lastEventId !== undefined) {
patch.lastEventId = args.lastEventId;
}
await ctx.db.patch(args.agentRowId, patch);
const updated = await ctx.db.get(args.agentRowId);
if (!updated) {
throw new ConvexError(
"Conversation agent could not be read after update"
);
}
return updated;
},
});
export const getForProject = internalQuery({
args: { projectId: v.id("projects") },
handler: async (ctx, args): Promise<Doc<"conversationAgents"> | null> =>
await ctx.db
.query("conversationAgents")
.withIndex("by_projectId", (q) => q.eq("projectId", args.projectId))
.unique(),
});
+3
View File
@@ -4,6 +4,9 @@ import { v } from "convex/values";
const app = defineApp({
env: {
AGENT_BACKEND_URL: v.optional(v.string()),
FLUE_DB_TOKEN: v.optional(v.string()),
FLUE_URL: v.optional(v.string()),
GITEA_TOKEN: v.optional(v.string()),
GITEA_URL: v.optional(v.string()),
GITEA_WEBHOOK_SECRET: v.optional(v.string()),
+156
View File
@@ -0,0 +1,156 @@
import { v } from "convex/values";
import type { Id } from "./_generated/dataModel";
import type { MutationCtx } from "./_generated/server";
import { internalMutation } from "./_generated/server";
/**
* The system actors used by the coordinator and agent tooling. The existing
* `events.actor` union already permits `{ kind: "system", service }` and
* `{ kind: "tool", tool }`; these constants name the canonical services.
*/
export const SYSTEM_SERVICE_PROJECT_SETUP = "project-setup";
export const SYSTEM_SERVICE_AGENT_GATEWAY = "agent-gateway";
interface AppendInput {
readonly actor:
| { readonly kind: "system"; readonly service: string }
| { readonly kind: "tool"; readonly tool: string };
readonly artifactIds?: readonly Id<"artifacts">[];
readonly causationId?: Id<"events">;
readonly correlationId: string;
readonly idempotencyKey: string;
readonly organizationId: Id<"organizations">;
readonly payload: unknown;
readonly projectId?: Id<"projects">;
readonly replyToEventId?: Id<"events">;
readonly scopeKind: "global" | "work";
readonly threadId?: Id<"threads">;
readonly type: string;
readonly visibility: "timeline" | "compact" | "internal";
readonly workId?: Id<"works">;
}
/**
* Append an immutable system/tool event with exact-once semantics: an event
* with the same `(organizationId, idempotencyKey)` is returned unchanged on
* retry. This is the shared primitive every coordinator phase and agent tool
* writes through.
*/
const appendEventOnce = async (
ctx: MutationCtx,
input: AppendInput
): Promise<Id<"events">> => {
const existing = await ctx.db
.query("events")
.withIndex("by_organizationId_and_idempotencyKey", (q) =>
q
.eq("organizationId", input.organizationId)
.eq("idempotencyKey", input.idempotencyKey)
)
.unique();
if (existing) {
return existing._id;
}
const now = Date.now();
return await ctx.db.insert("events", {
actor: input.actor,
artifactIds: input.artifactIds ? [...input.artifactIds] : undefined,
causationId: input.causationId,
correlationId: input.correlationId,
idempotencyKey: input.idempotencyKey,
occurredAt: now,
organizationId: input.organizationId,
payload: input.payload,
projectId: input.projectId,
recordedAt: now,
replyToEventId: input.replyToEventId,
scopeKind: input.scopeKind,
threadId: input.threadId,
type: input.type,
visibility: input.visibility,
workId: input.workId,
});
};
export const appendSystemEvent = internalMutation({
args: {
actorService: v.string(),
artifactIds: v.optional(v.array(v.id("artifacts"))),
causationId: v.optional(v.id("events")),
correlationId: v.string(),
idempotencyKey: v.string(),
organizationId: v.id("organizations"),
payload: v.any(),
projectId: v.optional(v.id("projects")),
replyToEventId: v.optional(v.id("events")),
scopeKind: v.union(v.literal("global"), v.literal("work")),
threadId: v.optional(v.id("threads")),
type: v.string(),
visibility: v.union(
v.literal("timeline"),
v.literal("compact"),
v.literal("internal")
),
workId: v.optional(v.id("works")),
},
handler: (ctx, args): Promise<Id<"events">> =>
appendEventOnce(ctx, {
actor: { kind: "system", service: args.actorService },
artifactIds: args.artifactIds,
causationId: args.causationId,
correlationId: args.correlationId,
idempotencyKey: args.idempotencyKey,
organizationId: args.organizationId,
payload: args.payload,
projectId: args.projectId,
replyToEventId: args.replyToEventId,
scopeKind: args.scopeKind,
threadId: args.threadId,
type: args.type,
visibility: args.visibility,
workId: args.workId,
}),
});
export const appendToolEvent = internalMutation({
args: {
actorTool: v.string(),
artifactIds: v.optional(v.array(v.id("artifacts"))),
causationId: v.optional(v.id("events")),
correlationId: v.string(),
idempotencyKey: v.string(),
organizationId: v.id("organizations"),
payload: v.any(),
projectId: v.optional(v.id("projects")),
replyToEventId: v.optional(v.id("events")),
scopeKind: v.union(v.literal("global"), v.literal("work")),
threadId: v.optional(v.id("threads")),
type: v.string(),
visibility: v.union(
v.literal("timeline"),
v.literal("compact"),
v.literal("internal")
),
workId: v.optional(v.id("works")),
},
handler: (ctx, args): Promise<Id<"events">> =>
appendEventOnce(ctx, {
actor: { kind: "tool", tool: args.actorTool },
artifactIds: args.artifactIds,
causationId: args.causationId,
correlationId: args.correlationId,
idempotencyKey: args.idempotencyKey,
organizationId: args.organizationId,
payload: args.payload,
projectId: args.projectId,
replyToEventId: args.replyToEventId,
scopeKind: args.scopeKind,
threadId: args.threadId,
type: args.type,
visibility: args.visibility,
workId: args.workId,
}),
});
export { appendEventOnce };
+121
View File
@@ -0,0 +1,121 @@
import { ConvexError, v } from "convex/values";
import type { Doc } from "./_generated/dataModel";
import { internalMutation, internalQuery } from "./_generated/server";
const flowRunStatus = v.union(
v.literal("running"),
v.literal("completed"),
v.literal("failed"),
v.literal("cancelled")
);
/**
* Create a FlowRun keyed idempotently by its idempotencyKey
* (`event:<eventId>:flow:<flowType>:<flowVersion>`). Retries reuse the existing
* run rather than creating duplicate specialist executions.
*/
export const create = internalMutation({
args: {
agentId: v.optional(v.string()),
flowType: v.string(),
flowVersion: v.string(),
idempotencyKey: v.string(),
organizationId: v.id("organizations"),
projectId: v.optional(v.id("projects")),
sourceEventId: v.id("events"),
threadId: v.optional(v.id("threads")),
workId: v.optional(v.id("works")),
},
handler: async (ctx, args): Promise<Doc<"flowRuns">> => {
// Idempotency: look up an existing run for the same source event + flow type.
const existing = await ctx.db
.query("flowRuns")
.withIndex("by_organizationId_and_source_event", (q) =>
q
.eq("organizationId", args.organizationId)
.eq("sourceEventId", args.sourceEventId)
)
.filter((q) => q.eq(q.field("flowType"), args.flowType))
.unique();
if (existing) {
return existing;
}
const timestamp = Date.now();
const runId = await ctx.db.insert("flowRuns", {
agentId: args.agentId,
attempt: 1,
flowType: args.flowType,
flowVersion: args.flowVersion,
idempotencyKey: args.idempotencyKey,
organizationId: args.organizationId,
projectId: args.projectId,
sourceEventId: args.sourceEventId,
startedAt: timestamp,
state: {},
status: "running",
threadId: args.threadId,
updatedAt: timestamp,
workId: args.workId,
});
const run = await ctx.db.get(runId);
if (!run) {
throw new ConvexError("Flow run could not be read after insert");
}
return run;
},
});
/**
* Patch a FlowRun's state/status. A terminal run cannot be revived by update.
*/
export const update = internalMutation({
args: {
finishedAt: v.optional(v.number()),
flowRunId: v.id("flowRuns"),
lastError: v.optional(v.string()),
state: v.optional(v.any()),
status: v.optional(flowRunStatus),
},
handler: async (ctx, args): Promise<Doc<"flowRuns">> => {
const run = await ctx.db.get(args.flowRunId);
if (!run) {
throw new ConvexError("Flow run not found");
}
// Terminal runs are immutable.
if (
run.status === "completed" ||
run.status === "cancelled" ||
run.status === "failed"
) {
return run;
}
const patch: Record<string, unknown> = { updatedAt: Date.now() };
if (args.state !== undefined) {
patch.state = args.state;
}
if (args.status !== undefined) {
patch.status = args.status;
}
if (args.finishedAt !== undefined) {
patch.finishedAt = args.finishedAt;
}
if (args.lastError !== undefined) {
patch.lastError = args.lastError;
}
await ctx.db.patch(args.flowRunId, patch);
const updated = await ctx.db.get(args.flowRunId);
if (!updated) {
throw new ConvexError("Flow run could not be read after update");
}
return updated;
},
});
export const get = internalQuery({
args: { flowRunId: v.id("flowRuns") },
handler: async (ctx, args): Promise<Doc<"flowRuns"> | null> =>
await ctx.db.get(args.flowRunId),
});
export { flowRunStatus };
+23
View File
@@ -1,5 +1,10 @@
import { httpRouter } from "convex/server";
import {
generateAgentUploadUrl,
postAgentMessage,
publishAgentArtifact,
} from "./agentHttp";
import { authComponent, createAuth } from "./auth";
import { githubWebhook, puterWebhook } from "./gitWebhooks";
@@ -19,4 +24,22 @@ http.route({
path: "/api/git/webhooks/puter",
});
http.route({
handler: postAgentMessage,
method: "POST",
path: "/api/agents/events/message",
});
http.route({
handler: publishAgentArtifact,
method: "POST",
path: "/api/agents/events/artifact",
});
http.route({
handler: generateAgentUploadUrl,
method: "POST",
path: "/api/agents/storage/generate-upload-url",
});
export default http;
@@ -1,7 +1,7 @@
import { ConvexError, v } from "convex/values";
import type { Doc, Id } from "./_generated/dataModel";
import { mutation, query } from "./_generated/server";
import { internalMutation, mutation, query } from "./_generated/server";
import { requireProjectMember } from "./authz";
const category = v.union(
@@ -82,3 +82,52 @@ export const upsert = mutation({
return saved;
},
});
/**
* Replace the entire detected environment-variable manifest for a Project in a
* single idempotent transaction. Called by the setup coordinator after it
* scans the root `.env.example` (names only — example/default text is never
* treated as a secret, stored, or displayed).
*
* Semantics:
* - All previously `detected` rows for the project are removed.
* - One new `detected` row is inserted per scanned name, category `optional`,
* status `missing` (configuration is not part of the readiness predicate).
* - `user`/`system` rows are untouched.
*
* Idempotency: re-running the coordinator with the same names produces the
* same final set of detected rows. No values are persisted.
*/
export const replaceDetectedManifest = internalMutation({
args: {
names: v.array(v.string()),
organizationId: v.id("organizations"),
projectId: v.id("projects"),
},
handler: async (ctx, args): Promise<{ count: number }> => {
const detected = await ctx.db
.query("projectEnvironmentVariables")
.withIndex("by_projectId", (q) => q.eq("projectId", args.projectId))
.filter((q) => q.eq(q.field("source"), "detected"))
.collect();
for (const row of detected) {
await ctx.db.delete(row._id);
}
const timestamp = Date.now();
const deduped = [...new Set(args.names.map((name) => name.trim())).values()]
.filter((name) => name.length > 0)
.sort();
for (const name of deduped) {
await ctx.db.insert("projectEnvironmentVariables", {
category: "optional",
name,
organizationId: args.organizationId,
projectId: args.projectId,
source: "detected",
status: "missing",
updatedAt: timestamp,
});
}
return { count: deduped.length };
},
});
+133
View File
@@ -0,0 +1,133 @@
import { ConvexError, v } from "convex/values";
import type { Doc, Id } from "./_generated/dataModel";
import { internalMutation, internalQuery } from "./_generated/server";
const RUNTIME_STATUS = [
"requested",
"creating_vm",
"cloning",
"checking_repository",
"ready",
"failed",
] as const;
const runtimeStatus = v.union(
v.literal("requested"),
v.literal("creating_vm"),
v.literal("cloning"),
v.literal("checking_repository"),
v.literal("ready"),
v.literal("failed")
);
/**
* Create or reuse a ProjectRuntime row, keyed idempotently by projectId.
* Idempotency key: `project:<projectId>:vm:v1` (per the shared contract).
* Re-runs (retries) return the existing row without duplicating it.
*/
export const requestRuntime = internalMutation({
args: {
idempotencyKey: v.string(),
organizationId: v.id("organizations"),
projectId: v.id("projects"),
provider: v.string(),
},
handler: async (ctx, args): Promise<Id<"projectRuntimes">> => {
const existing = await ctx.db
.query("projectRuntimes")
.withIndex("by_projectId", (q) => q.eq("projectId", args.projectId))
.unique();
if (existing) {
return existing._id;
}
const timestamp = Date.now();
return await ctx.db.insert("projectRuntimes", {
createdAt: timestamp,
idempotencyKey: args.idempotencyKey,
organizationId: args.organizationId,
projectId: args.projectId,
provider: args.provider,
status: "requested",
updatedAt: timestamp,
});
},
});
/**
* Advance a ProjectRuntime to a new phase. A terminal "ready" or "failed"
* status cannot be overwritten by an earlier phase on retry, so a completed
* setup is never partially rewound. Idempotent: re-applying the same status
* is a no-op.
*/
export const markStatus = internalMutation({
args: {
lastError: v.optional(v.string()),
repositoryCommit: v.optional(v.string()),
repositoryPath: v.optional(v.string()),
runtimeId: v.optional(v.string()),
runtimeRowId: v.id("projectRuntimes"),
status: runtimeStatus,
vmId: v.optional(v.string()),
workspacePath: v.optional(v.string()),
},
handler: async (ctx, args): Promise<Doc<"projectRuntimes">> => {
const runtime = await ctx.db.get(args.runtimeRowId);
if (!runtime) {
throw new ConvexError("Project runtime not found");
}
// Never rewind a terminal state. A retry that re-enters the coordinator
// after success must observe readiness, not re-run phases.
if (
(runtime.status === "ready" || runtime.status === "failed") &&
args.status !== runtime.status
) {
return runtime;
}
const patch: Record<string, unknown> = {
status: args.status,
updatedAt: Date.now(),
};
if (args.runtimeId !== undefined) {
patch.runtimeId = args.runtimeId;
}
if (args.vmId !== undefined) {
patch.vmId = args.vmId;
}
if (args.workspacePath !== undefined) {
patch.workspacePath = args.workspacePath;
}
if (args.repositoryPath !== undefined) {
patch.repositoryPath = args.repositoryPath;
}
if (args.repositoryCommit !== undefined) {
patch.repositoryCommit = args.repositoryCommit;
}
if (args.lastError !== undefined) {
patch.lastError = args.lastError;
}
await ctx.db.patch(args.runtimeRowId, patch);
const updated = await ctx.db.get(args.runtimeRowId);
if (!updated) {
throw new ConvexError("Project runtime could not be read after update");
}
return updated;
},
});
export const getForProject = internalQuery({
args: { projectId: v.id("projects") },
handler: async (ctx, args): Promise<Doc<"projectRuntimes"> | null> =>
await ctx.db
.query("projectRuntimes")
.withIndex("by_projectId", (q) => q.eq("projectId", args.projectId))
.unique(),
});
export const get = internalQuery({
args: { runtimeRowId: v.id("projectRuntimes") },
handler: async (ctx, args): Promise<Doc<"projectRuntimes"> | null> =>
await ctx.db.get(args.runtimeRowId),
});
export { RUNTIME_STATUS };
+528
View File
@@ -0,0 +1,528 @@
"use node";
import { env } from "@code/env/convex";
import { makeFunctionReference } from "convex/server";
import { ConvexError, v } from "convex/values";
import type { Id } from "./_generated/dataModel";
import { internalAction } from "./_generated/server";
import type { ActionCtx } from "./_generated/server";
// Idempotency keys (shared contract):
const runtimeIdempotencyKey = (projectId: Id<"projects">) =>
`project:${projectId}:vm:v1`;
const readyEventIdempotencyKey = (projectId: Id<"projects">) =>
`project:${projectId}:ready:v1`;
const dispatchIdempotencyKey = (eventId: Id<"events">, agentId: string) =>
`event:${eventId}:agent:${agentId}:handler:v1`;
const lifecycleEventIdempotencyKey = (
projectId: Id<"projects">,
phase: string
) => `project:${projectId}:setup:${phase}:v1`;
const PROJECT_AGENT_TYPE = "project";
const requestRuntimeRef = makeFunctionReference<
"mutation",
{
idempotencyKey: string;
organizationId: Id<"organizations">;
projectId: Id<"projects">;
provider: string;
},
Id<"projectRuntimes">
>("projectRuntimes:requestRuntime");
const markRuntimeStatusRef = makeFunctionReference<
"mutation",
{
runtimeRowId: Id<"projectRuntimes">;
status: string;
runtimeId?: string;
vmId?: string;
workspacePath?: string;
repositoryPath?: string;
repositoryCommit?: string;
lastError?: string;
},
unknown
>("projectRuntimes:markStatus");
const getRuntimeRef = makeFunctionReference<
"query",
{ projectId: Id<"projects"> },
unknown
>("projectRuntimes:getForProject");
const replaceDetectedManifestRef = makeFunctionReference<
"mutation",
{
names: string[];
organizationId: Id<"organizations">;
projectId: Id<"projects">;
},
{ count: number }
>("projectEnvironmentVariables:replaceDetectedManifest");
const registerAgentRef = makeFunctionReference<
"mutation",
{
agentType: string;
organizationId: Id<"organizations">;
projectId: Id<"projects">;
runtimeId?: Id<"projectRuntimes">;
},
{ _id: Id<"conversationAgents">; id: string }
>("conversationAgents:register");
const markAgentStatusRef = makeFunctionReference<
"mutation",
{ agentRowId: Id<"conversationAgents">; status: string },
unknown
>("conversationAgents:markStatus");
const appendSystemEventRef = makeFunctionReference<
"mutation",
{
actorService: string;
correlationId: string;
idempotencyKey: string;
organizationId: Id<"organizations">;
payload: unknown;
projectId: Id<"projects">;
scopeKind: "global" | "work";
type: string;
visibility: "timeline" | "compact" | "internal";
},
Id<"events">
>("events:appendSystemEvent");
const createDispatchRef = makeFunctionReference<
"mutation",
{
agentId: string;
idempotencyKey: string;
organizationId: Id<"organizations">;
projectId: Id<"projects">;
sourceEventId: Id<"events">;
},
{ created: boolean; dispatch: { _id: Id<"agentDispatches">; status: string } }
>("agentDispatches:create");
const markDispatchSendingRef = makeFunctionReference<
"mutation",
{ dispatchId: Id<"agentDispatches"> },
unknown
>("agentDispatches:markSending");
const markDispatchAcceptedRef = makeFunctionReference<
"mutation",
{ dispatchId: Id<"agentDispatches"> },
unknown
>("agentDispatches:markAccepted");
const markDispatchCompletedRef = makeFunctionReference<
"mutation",
{ dispatchId: Id<"agentDispatches"> },
unknown
>("agentDispatches:markCompleted");
const markDispatchFailedRef = makeFunctionReference<
"mutation",
{ dispatchId: Id<"agentDispatches">; error: string; retry: boolean },
unknown
>("agentDispatches:markFailed");
const getProjectSourceRef = makeFunctionReference<
"query",
{ projectId: Id<"projects"> },
{
organizationId: Id<"organizations">;
sourceUrl: string;
defaultBranch: string | null;
name: string;
} | null
>("projectSetupQueries:getProjectSource");
/** Resolve the agent backend URL. AGENT_BACKEND_URL takes precedence over FLUE_URL. */
const backendUrl = (): string => {
const url = env.AGENT_BACKEND_URL ?? env.FLUE_URL;
if (!url) {
throw new ConvexError(
"AGENT_BACKEND_URL or FLUE_URL must be configured for project setup"
);
}
return url;
};
const authHeaders = (): Record<string, string> => {
if (!env.FLUE_DB_TOKEN) {
throw new ConvexError("FLUE_DB_TOKEN must be configured for project setup");
}
return {
authorization: `Bearer ${env.FLUE_DB_TOKEN}`,
"content-type": "application/json",
};
};
/** Wire shape of the setup-coordinator result returned by the Hono endpoint. */
interface SetupRuntimeResult {
runtimeId?: string;
vmId?: string;
workspacePath?: string;
repositoryPath?: string;
repositoryCommit?: string;
environmentVariableNames?: string[];
}
interface RuntimeRow {
_id: Id<"projectRuntimes">;
status: string;
organizationId: Id<"organizations">;
}
const asRuntimeRow = (doc: unknown): RuntimeRow => {
const row = doc as Partial<RuntimeRow> | null;
if (!row || !row._id || !row.organizationId) {
throw new ConvexError("Project runtime row is malformed");
}
return {
_id: row._id,
organizationId: row.organizationId,
status: row.status ?? "requested",
};
};
interface DispatchReadyArgs {
agentId: string;
correlationId: string;
organizationId: Id<"organizations">;
projectId: Id<"projects">;
readyEventId: Id<"events">;
sourceEvent: { type: string; payload: unknown };
}
const dispatchReadyWithEvent = async (
ctx: ActionCtx,
args: DispatchReadyArgs
): Promise<void> => {
const created = (await ctx.runMutation(createDispatchRef, {
agentId: args.agentId,
idempotencyKey: dispatchIdempotencyKey(args.readyEventId, args.agentId),
organizationId: args.organizationId,
projectId: args.projectId,
sourceEventId: args.readyEventId,
})) as {
created: boolean;
dispatch: { _id: Id<"agentDispatches">; status: string };
};
const dispatchId = created.dispatch._id;
// An already-completed dispatch from a prior run must never be re-sent; the
// idempotency key guarantees this is the same logical delivery.
if (created.dispatch.status === "completed") {
return;
}
await ctx.runMutation(markDispatchSendingRef, { dispatchId });
try {
const response = await fetch(
`${backendUrl()}/internal/agents/${encodeURIComponent(args.agentId)}/events`,
{
body: JSON.stringify({
agentId: args.agentId,
correlationId: args.correlationId,
dispatchId,
organizationId: args.organizationId,
projectId: args.projectId,
sourceEvent: args.sourceEvent,
sourceEventId: args.readyEventId,
}),
headers: authHeaders(),
method: "POST",
}
);
if (response.status === 202 || response.status === 200) {
await ctx.runMutation(markDispatchAcceptedRef, { dispatchId });
await ctx.runMutation(markDispatchCompletedRef, { dispatchId });
return;
}
const text = await response.text().catch(() => "");
throw new Error(
`Agent dispatch failed: ${response.status} ${response.statusText}${text ? ` — ${text}` : ""}`
);
} catch (error) {
const message = error instanceof Error ? error.message : String(error);
await ctx.runMutation(markDispatchFailedRef, {
dispatchId,
error: message,
retry: true,
});
throw error;
}
};
const callSetupEndpoint = async (args: {
branch: string;
projectId: Id<"projects">;
repositoryUrl: string;
}): Promise<SetupRuntimeResult> => {
const response = await fetch(`${backendUrl()}/internal/project-setup`, {
body: JSON.stringify({
branch: args.branch,
projectId: args.projectId,
repositoryUrl: args.repositoryUrl,
}),
headers: authHeaders(),
method: "POST",
});
if (!response.ok) {
const text = await response.text().catch(() => "");
throw new Error(
`Project setup endpoint returned ${response.status} ${response.statusText}${text ? ` — ${text}` : ""}`
);
}
const payload = (await response.json()) as unknown;
if (typeof payload !== "object" || payload === null) {
throw new Error("Project setup endpoint returned a malformed response");
}
return payload as SetupRuntimeResult;
};
/**
* The durable project setup coordinator. Drives one Project through:
* requested -> creating_vm -> cloning -> checking_repository -> ready
* (or -> failed).
*
* It calls a single Hono internal setup endpoint that performs the AgentOS VM
* get-or-create, shallow clone, readability verification, and `.env.example`
* scan, returning the runtime facts. Each Convex-visible phase transition is
* persisted to `projectRuntimes` so retries and the setup screen observe
* progress. On success it registers the conversation agent, appends the
* exact-once `project.ready` event, creates the `agentDispatches` row, and
* dispatches the ready event to the Hono `POST /internal/agents/:agentId/events`
* endpoint.
*
* This function never mutates the repository and never persists example/env
* values. Idempotency keys guarantee retries create no duplicate VMs, agents,
* events, or dispatches.
*
* `correlationId` ties all setup events together; `agentId` is the stable
* `conversation:<org>:<project>:v1` agent identity.
*/
export const runSetup = internalAction({
args: {
agentId: v.string(),
correlationId: v.string(),
organizationId: v.id("organizations"),
projectId: v.id("projects"),
},
handler: async (ctx, args) => {
const source = await ctx.runQuery(getProjectSourceRef, {
projectId: args.projectId,
});
if (!source || source.organizationId !== args.organizationId) {
throw new ConvexError("Project not found within organization");
}
// Request (or reuse) the runtime row. Terminal-safe: a ready/failed row is
// returned as-is by the mutation.
const runtimeRowId = await ctx.runMutation(requestRuntimeRef, {
idempotencyKey: runtimeIdempotencyKey(args.projectId),
organizationId: args.organizationId,
projectId: args.projectId,
provider: "agentos",
});
// Re-check terminal state: if already ready, just (re)dispatch ready.
const current = asRuntimeRow(
await ctx.runQuery(getRuntimeRef, {
projectId: args.projectId,
})
);
if (current.status === "ready") {
// Terminal `ready` is only reached after the exact-once `project.ready`
// event was appended AND its dispatch completed on a prior run, so a
// re-entry simply reports readiness without re-running phases or
// re-delivering (the dispatch idempotency key would no-op anyway).
return {
projectId: args.projectId,
runtimeId: runtimeRowId,
status: "ready",
};
}
if (current.status === "failed") {
return {
projectId: args.projectId,
runtimeId: runtimeRowId,
status: "failed",
};
}
const branch = source.defaultBranch ?? "main";
// --- Phase: creating_vm ---
await ctx.runMutation(markRuntimeStatusRef, {
runtimeRowId,
status: "creating_vm",
});
await ctx.runMutation(appendSystemEventRef, {
actorService: "project-setup",
correlationId: args.correlationId,
idempotencyKey: lifecycleEventIdempotencyKey(
args.projectId,
"creating_vm"
),
organizationId: args.organizationId,
payload: {},
projectId: args.projectId,
scopeKind: "global",
type: "project.setup.creating_vm",
visibility: "internal",
});
// --- Phase: cloning + checking_repository + env scan (single Hono call) ---
await ctx.runMutation(markRuntimeStatusRef, {
runtimeRowId,
status: "cloning",
});
await ctx.runMutation(appendSystemEventRef, {
actorService: "project-setup",
correlationId: args.correlationId,
idempotencyKey: lifecycleEventIdempotencyKey(args.projectId, "cloning"),
organizationId: args.organizationId,
payload: { branch },
projectId: args.projectId,
scopeKind: "global",
type: "project.setup.cloning",
visibility: "internal",
});
let result: SetupRuntimeResult;
try {
result = await callSetupEndpoint({
branch,
projectId: args.projectId,
repositoryUrl: source.sourceUrl,
});
} catch (error) {
const message = error instanceof Error ? error.message : String(error);
await ctx.runMutation(markRuntimeStatusRef, {
lastError: message,
runtimeRowId,
status: "failed",
});
await ctx.runMutation(appendSystemEventRef, {
actorService: "project-setup",
correlationId: args.correlationId,
idempotencyKey: `${readyEventIdempotencyKey(args.projectId)}:failed`,
organizationId: args.organizationId,
payload: { error: message },
projectId: args.projectId,
scopeKind: "global",
type: "project.setup.failed",
visibility: "internal",
});
return {
error: message,
projectId: args.projectId,
runtimeId: runtimeRowId,
status: "failed",
};
}
// --- Phase: checking_repository (persisted) ---
await ctx.runMutation(markRuntimeStatusRef, {
repositoryCommit: result.repositoryCommit,
repositoryPath: result.repositoryPath,
runtimeId: result.runtimeId,
runtimeRowId,
status: "checking_repository",
vmId: result.vmId,
workspacePath: result.workspacePath,
});
await ctx.runMutation(appendSystemEventRef, {
actorService: "project-setup",
correlationId: args.correlationId,
idempotencyKey: lifecycleEventIdempotencyKey(
args.projectId,
"checking_repository"
),
organizationId: args.organizationId,
payload: {
repositoryCommit: result.repositoryCommit ?? null,
runtimeId: result.runtimeId ?? null,
vmId: result.vmId ?? null,
},
projectId: args.projectId,
scopeKind: "global",
type: "project.setup.checking_repository",
visibility: "internal",
});
// Persist the detected env-name manifest (names only; non-blocking).
const envNames = result.environmentVariableNames ?? [];
await ctx.runMutation(replaceDetectedManifestRef, {
names: envNames,
organizationId: args.organizationId,
projectId: args.projectId,
});
// Register the one project-bound conversation agent (idempotent).
const agent = (await ctx.runMutation(registerAgentRef, {
agentType: PROJECT_AGENT_TYPE,
organizationId: args.organizationId,
projectId: args.projectId,
runtimeId: runtimeRowId,
})) as { _id: Id<"conversationAgents">; id: string };
await ctx.runMutation(markAgentStatusRef, {
agentRowId: agent._id,
status: "active",
});
// --- Phase: ready ---
// Append the exact-once `project.ready` event (the durable onboarding
// trigger) and deliver it BEFORE flipping the runtime terminal. A delivery
// failure must throw so Convex retries the coordinator; the runtime stays
// non-terminal, and the idempotency keys make the retry reuse the same
// event/dispatch rather than duplicating it.
const readyEventId = await ctx.runMutation(appendSystemEventRef, {
actorService: "project-setup",
correlationId: args.correlationId,
idempotencyKey: readyEventIdempotencyKey(args.projectId),
organizationId: args.organizationId,
payload: {
agentId: args.agentId,
environmentVariableNames: envNames,
repositoryCommit: result.repositoryCommit ?? null,
runtimeId: result.runtimeId ?? null,
},
projectId: args.projectId,
scopeKind: "global",
type: "project.ready",
visibility: "timeline",
});
// Dispatch the ready event to the project conversation agent. Throws on
// delivery failure; only success returns, after which `ready` is durable.
await dispatchReadyWithEvent(ctx, {
agentId: args.agentId,
correlationId: args.correlationId,
organizationId: args.organizationId,
projectId: args.projectId,
readyEventId,
sourceEvent: {
payload: {
agentId: args.agentId,
environmentVariableNames: envNames,
repositoryCommit: result.repositoryCommit ?? null,
runtimeId: result.runtimeId ?? null,
},
type: "project.ready",
},
});
// Runtime becomes terminal only after the ready event is durably delivered.
await ctx.runMutation(markRuntimeStatusRef, {
runtimeRowId,
status: "ready",
});
return {
projectId: args.projectId,
readyEventId,
runtimeId: runtimeRowId,
status: "ready",
};
},
});
export { runtimeIdempotencyKey, readyEventIdempotencyKey };
@@ -0,0 +1,64 @@
import { v } from "convex/values";
import { internalQuery } from "./_generated/server";
/**
* Query the project source metadata the coordinator needs (clone URL, default
* branch, org). Internal so the Node action can read it without touching the DB.
*
* Lives outside the `"use node"` projectSetup module because Convex forbids
* query/mutation definitions in Node-runtime modules.
*/
export const getProjectSource = internalQuery({
args: { projectId: v.id("projects") },
handler: async (ctx, args) => {
const project = await ctx.db.get(args.projectId);
if (!project) {
return null;
}
return {
defaultBranch: project.defaultBranch ?? null,
name: project.name,
organizationId: project.organizationId,
sourceUrl: project.sourceUrl,
};
},
});
/**
* Read-only project setup status for the setup screen: latest runtime row plus
* the setup-scoped events. Auth-gated to project members.
*/
export const getStatus = internalQuery({
args: { projectId: v.id("projects") },
handler: async (ctx, args) => {
const project = await ctx.db.get(args.projectId);
if (!project) {
return { events: [], runtime: null };
}
const runtime = await ctx.db
.query("projectRuntimes")
.withIndex("by_projectId", (q) => q.eq("projectId", args.projectId))
.unique();
const events = await ctx.db
.query("events")
.withIndex("by_organizationId_and_projectId_and_recordedAt", (q) =>
q
.eq("organizationId", project.organizationId)
.eq("projectId", args.projectId)
)
.order("desc")
.filter((q) =>
q.or(
q.eq(q.field("type"), "project.setup.creating_vm"),
q.eq(q.field("type"), "project.setup.cloning"),
q.eq(q.field("type"), "project.setup.checking_repository"),
q.eq(q.field("type"), "project.ready"),
q.eq(q.field("type"), "project.setup.failed"),
q.eq(q.field("type"), "project.setup.started")
)
)
.take(50);
return { events, runtime };
},
});
+52 -8
View File
@@ -7,6 +7,7 @@ import type {
ProjectImportOutcome,
ProjectView,
} from "@code/primitives/project";
import { makeFunctionReference } from "convex/server";
import { ConvexError, v } from "convex/values";
import { Effect } from "effect";
@@ -18,8 +19,20 @@ import {
requireCurrentOrganization,
requireProjectMember,
} from "./authz";
import { conversationAgentId } from "./conversationAgents";
import { inspectPublicGit } from "./publicGit";
const runSetupRef = makeFunctionReference<
"action",
{
agentId: string;
correlationId: string;
organizationId: Id<"organizations">;
projectId: Id<"projects">;
},
unknown
>("projectSetup:runSetup");
const toProjectView = async (
ctx: Parameters<typeof requireCurrentOrganization>[0],
project: Doc<"projects">
@@ -197,6 +210,13 @@ export const getSetup = query({
.order("desc")
.filter((q) =>
q.or(
// Lifecycle event types emitted by the project-setup coordinator.
q.eq(q.field("type"), "project.setup.creating_vm"),
q.eq(q.field("type"), "project.setup.cloning"),
q.eq(q.field("type"), "project.setup.checking_repository"),
q.eq(q.field("type"), "project.ready"),
q.eq(q.field("type"), "project.setup.failed"),
// Legacy / environment / preview event types (compatible).
q.eq(q.field("type"), "project.setup.started"),
q.eq(q.field("type"), "project.setup.environment_required"),
q.eq(q.field("type"), "project.setup.environment_updated"),
@@ -233,6 +253,13 @@ export const appendSetupEvent = internalMutation({
payload: v.any(),
projectId: v.id("projects"),
type: v.union(
// Lifecycle event types emitted by the project-setup coordinator.
v.literal("project.setup.creating_vm"),
v.literal("project.setup.cloning"),
v.literal("project.setup.checking_repository"),
v.literal("project.ready"),
v.literal("project.setup.failed"),
// Legacy / environment / preview event types (compatible).
v.literal("project.setup.started"),
v.literal("project.setup.environment_required"),
v.literal("project.setup.environment_updated"),
@@ -290,15 +317,32 @@ export const importPublicGit = action({
const remote = await Effect.runPromise(
decodePublicGitImportResult(await inspectPublicGit(source))
);
return await ctx.runMutation(internal.projects.persistPublicGitImport, {
remote: {
defaultBranch: remote.defaultBranch,
documents: remote.documents.map((document) => ({ ...document })),
warnings: remote.warnings.map((warning) => ({ ...warning })),
},
source,
userId,
const outcome = await ctx.runMutation(
internal.projects.persistPublicGitImport,
{
remote: {
defaultBranch: remote.defaultBranch,
documents: remote.documents.map((document) => ({ ...document })),
warnings: remote.warnings.map((warning) => ({ ...warning })),
},
source,
userId,
}
);
// Kick off the idempotent project setup coordinator (AgentOS VM + clone +
// readiness + project.ready + agent dispatch). Scheduled as an internal
// action so retries are safe: every step is keyed idempotently.
const projectId = outcome.id as unknown as Id<"projects">;
const organizationId =
outcome.organizationId as unknown as Id<"organizations">;
const agentId = conversationAgentId(organizationId, projectId);
await ctx.scheduler.runAfter(0, runSetupRef, {
agentId,
correlationId: `import:${projectId}`,
organizationId,
projectId,
});
return outcome;
},
});
+155
View File
@@ -365,6 +365,7 @@ export default defineSchema({
threadId: v.optional(v.id("threads")),
title: v.string(),
version: v.number(),
storageId: v.optional(v.id("_storage")),
workId: v.optional(v.id("works")),
})
.index("by_projectId_and_createdAt", ["projectId", "createdAt"])
@@ -375,4 +376,158 @@ export default defineSchema({
"logicalKey",
"version",
]),
projectRuntimes: defineTable({
createdAt: v.number(),
idempotencyKey: v.string(),
lastError: v.optional(v.string()),
organizationId: v.id("organizations"),
projectId: v.id("projects"),
provider: v.string(),
repositoryCommit: v.optional(v.string()),
repositoryPath: v.optional(v.string()),
runtimeId: v.optional(v.string()),
status: v.union(
v.literal("requested"),
v.literal("creating_vm"),
v.literal("cloning"),
v.literal("checking_repository"),
v.literal("ready"),
v.literal("failed")
),
updatedAt: v.number(),
vmId: v.optional(v.string()),
workspacePath: v.optional(v.string()),
})
.index("by_projectId", ["projectId"])
.index("by_organizationId_and_status", ["organizationId", "status"])
.index("by_organizationId_and_runtime_id", ["organizationId", "runtimeId"]),
conversationAgents: defineTable({
agentType: v.string(),
createdAt: v.number(),
flueConversationId: v.optional(v.string()),
id: v.string(),
lastEventId: v.optional(v.id("events")),
organizationId: v.id("organizations"),
projectId: v.id("projects"),
runtimeId: v.optional(v.id("projectRuntimes")),
status: v.union(
v.literal("registered"),
v.literal("active"),
v.literal("idle"),
v.literal("disabled")
),
updatedAt: v.number(),
})
.index("by_projectId", ["projectId"])
.index("by_projectId_and_status", ["projectId", "status"])
.index("by_organizationId_and_status", ["organizationId", "status"]),
agentDispatches: defineTable({
agentId: v.string(),
attempt: v.number(),
createdAt: v.number(),
handlerVersion: v.string(),
idempotencyKey: v.string(),
lastError: v.optional(v.string()),
organizationId: v.id("organizations"),
projectId: v.optional(v.id("projects")),
sourceEventId: v.id("events"),
status: v.union(
v.literal("pending"),
v.literal("sending"),
v.literal("accepted"),
v.literal("completed"),
v.literal("failed")
),
threadId: v.optional(v.id("threads")),
updatedAt: v.number(),
workId: v.optional(v.id("works")),
})
.index("by_organizationId_and_source_event_handler", [
"organizationId",
"sourceEventId",
"agentId",
"handlerVersion",
])
.index("by_organizationId_and_status", ["organizationId", "status"])
.index("by_organizationId_and_status_and_updatedAt", [
"organizationId",
"status",
"updatedAt",
]),
flowRuns: defineTable({
agentId: v.optional(v.string()),
attempt: v.number(),
finishedAt: v.optional(v.number()),
flowType: v.string(),
flowVersion: v.string(),
idempotencyKey: v.string(),
organizationId: v.id("organizations"),
projectId: v.optional(v.id("projects")),
sourceEventId: v.id("events"),
startedAt: v.number(),
state: v.any(),
status: v.union(
v.literal("running"),
v.literal("completed"),
v.literal("failed"),
v.literal("cancelled")
),
threadId: v.optional(v.id("threads")),
updatedAt: v.number(),
workId: v.optional(v.id("works")),
})
.index("by_organizationId_and_thread", ["organizationId", "threadId"])
.index("by_organizationId_and_work_and_status", [
"organizationId",
"workId",
"status",
])
.index("by_organizationId_and_agent_and_status", [
"organizationId",
"agentId",
"status",
])
.index("by_organizationId_and_source_event", [
"organizationId",
"sourceEventId",
]),
artifactBuilds: defineTable({
artifactId: v.optional(v.id("artifacts")),
artifactKind: v.union(
v.literal("project_setup"),
v.literal("summary"),
v.literal("plan"),
v.literal("preview"),
v.literal("blocker"),
v.literal("report")
),
createdAt: v.number(),
flowRunId: v.optional(v.id("flowRuns")),
idempotencyKey: v.string(),
lastError: v.optional(v.string()),
organizationId: v.id("organizations"),
outputStorageId: v.optional(v.id("_storage")),
projectId: v.optional(v.id("projects")),
sourceManifest: v.any(),
status: v.union(
v.literal("pending"),
v.literal("building"),
v.literal("ready"),
v.literal("failed")
),
threadId: v.optional(v.id("threads")),
updatedAt: v.number(),
workId: v.optional(v.id("works")),
})
.index("by_organizationId_and_flow_run", ["organizationId", "flowRunId"])
.index("by_organizationId_and_status_and_updatedAt", [
"organizationId",
"status",
"updatedAt",
])
.index("by_organizationId_and_project_and_createdAt", [
"organizationId",
"projectId",
"createdAt",
]),
});
+1
View File
@@ -4,6 +4,7 @@
"private": true,
"type": "module",
"exports": {
"./agent": "./src/agent.ts",
"./convex": "./src/convex.ts",
"./native": "./src/native.ts",
"./server": "./src/server.ts",
+89
View File
@@ -0,0 +1,89 @@
import { createEnv } from "@t3-oss/env-core";
import { z } from "zod";
/**
* Runtime environment for the Flue 2.0 + Hono + AgentOS intelligence
* runtime (@code/agents). Validated once at process start.
*
* Convex remains the durable source of truth; these keys configure only the
* runtime process: model routing, persistence, service auth, and the
* AgentOS/Rivet host connection.
*/
export const env = createEnv({
emptyStringAsUndefined: true,
runtimeEnv: process.env,
server: {
AGENT_MODEL_API: z.string().min(1).default("openai-completions"),
AGENT_MODEL_API_KEY: z.string().min(1),
AGENT_MODEL_BASE_URL: z.url(),
AGENT_MODEL_CONTEXT_WINDOW: z.coerce
.number()
.int()
.positive()
.default(1_048_576),
AGENT_MODEL_MAX_TOKENS: z.coerce.number().int().positive().default(131_072),
AGENT_MODEL_NAME: z.string().min(1).default("mimo-v2.5"),
// Model routing — Cheaptricks OpenAI-compatible provider.
AGENT_MODEL_PROVIDER: z.string().min(1).default("cheaptricks"),
// Local workspace root for project VMs and generated artifacts.
AGENT_WORKSPACE_ROOT: z.string().min(1),
// Convex Site base for service HTTP actions (the .convex.site host).
// Runtime → Convex callbacks POST bearer-authenticated calls here.
CONVEX_SITE_URL: z.url(),
// Convex control plane (the .convex.cloud deployment URL).
CONVEX_URL: z.url(),
// Flue SQLite canonical conversation state.
FLUE_DB_PATH: z.string().min(1).default("/srv/zopu/data/flue/flue.db"),
// Service auth between Convex and this runtime.
FLUE_DB_TOKEN: z.string().min(1),
HOST: z.string().min(1).default("127.0.0.1"),
// HTTP listen config.
PORT: z.coerce.number().int().positive().default(3000),
// AgentOS / Rivet host connection. Accepts a full URL
// (http://namespace:token@127.0.0.1:6420) or a bare host:port
// (127.0.0.1:6420), normalizing the latter to http://.
RIVET_ENDPOINT: z
.string()
.min(1)
.refine(
(val) => {
try {
const candidate = /^https?:\/\//u.test(val) ? val : `http://${val}`;
void new URL(candidate);
return true;
} catch {
return false;
}
},
{ message: "RIVET_ENDPOINT must be a valid URL or host:port" }
)
.transform((val) => (/^https?:\/\//u.test(val) ? val : `http://${val}`)),
// Distinct endpoint the Rivet Engine invokes to reach this process's
// in-process serverless registry handler (mounted by the Hono app at
// RIVET_SERVERLESS_BASE_PATH). This is NOT the Engine control-plane URL
// (that is RIVET_ENDPOINT); it is how the Engine calls back into the
// registry handler co-hosted in this Node process. Defaults to the local
// app listener.
RIVET_SERVERLESS_ENDPOINT: z
.string()
.min(1)
.refine(
(val) => {
try {
const candidate = /^https?:\/\//u.test(val) ? val : `http://${val}`;
void new URL(candidate);
return true;
} catch {
return false;
}
},
{
message: "RIVET_SERVERLESS_ENDPOINT must be a valid URL or host:port",
}
)
.transform((val) => (/^https?:\/\//u.test(val) ? val : `http://${val}`))
.default("http://127.0.0.1:3000/internal/rivet"),
RIVET_WORKSPACE_TOKEN: z.string().min(1),
},
skipValidation: process.env.SKIP_ENV_VALIDATION === "true",
});
+3
View File
@@ -5,7 +5,10 @@ export const env = createEnv({
emptyStringAsUndefined: true,
runtimeEnv: process.env,
server: {
AGENT_BACKEND_URL: z.url().optional(),
CONVEX_SITE_URL: z.url().optional(),
FLUE_DB_TOKEN: z.string().min(1).optional(),
FLUE_URL: z.url().optional(),
GITEA_TOKEN: z.string().min(1).optional(),
GITEA_URL: z.url().default("https://git.openputer.com"),
GITEA_WEBHOOK_SECRET: z.string().min(1).optional(),