| 1 | import { requireAccount, type AppContext } from "../auth/principal.js"; |
| 2 | import type { Env } from "../env.js"; |
| 3 | import type { loadRepo } from "../http/common.js"; |
| 4 | import type { Account, Directory } from "../lib/directory.js"; |
| 5 | import { HttpError } from "../lib/util.js"; |
| 6 | import { atLeast, type Caller } from "../repo/types.js"; |
| 7 | import { mayDriveAgent, taskCaller, TaskAccessError } from "./access.js"; |
| 8 | import { ModelKeys, PROVIDERS, type Provider } from "./keys.js"; |
| 9 | import type { TaskRow, TaskView } from "./ProjectDO.js"; |
| 10 | import { isRuntime, MAX_COMMAND, MAX_GOAL, RUNTIMES, type RuntimeId, type TranscriptLine } from "./runtimes.js"; |
| 11 | |
| 12 | /** |
| 13 | * Who may do what with agent tasks. Shared by the REST API, MCP and the web. |
| 14 | * |
| 15 | * - Reading tasks and transcripts: write access to the repository or more. |
| 16 | * - Starting, sending a message and retrying: only a person, in a browser |
| 17 | * session (each run spends their model key), and for an existing task only |
| 18 | * the person who started it. No token, agent, job or sandbox can. |
| 19 | * - Stopping: the person who started it, its agent, or a repository admin. |
| 20 | */ |
| 21 | |
| 22 | type Loaded = Awaited<ReturnType<typeof loadRepo>>; |
| 23 | |
| 24 | export const projectStub = (env: Env, repoId: string) => env.PROJECT.get(env.PROJECT.idFromName(repoId)); |
| 25 | |
| 26 | /** Tasks are visible with write access or more: transcripts show an agent's reasoning and command output. */ |
| 27 | export function requireTaskReader(l: Loaded): void { |
| 28 | if (!atLeast(l.caller.role, "write")) throw new HttpError(404, "not found"); |
| 29 | } |
| 30 | |
| 31 | /** A person signed in in a browser (or the local dev sign-in): the only one who can spend their model key. */ |
| 32 | export function requirePersonSession(c: AppContext): Account { |
| 33 | const p = c.get("principal"); |
| 34 | if (p.type === "anonymous") throw new HttpError(401, "sign in required"); |
| 35 | if (p.type !== "account" || (p.via !== "session" && p.via !== "dev")) throw new HttpError(403, "starting or steering an agent task needs a browser session: each run spends your own model key, so no token can do it"); |
| 36 | const me = requireAccount(c); |
| 37 | if (me.kind !== "user") throw new HttpError(403, "only people can start agent tasks"); |
| 38 | return me; |
| 39 | } |
| 40 | |
| 41 | export function taskNumber(s: string | undefined): number { |
| 42 | const n = Number(s); |
| 43 | if (!Number.isInteger(n) || n <= 0) throw new HttpError(404, "task not found"); |
| 44 | return n; |
| 45 | } |
| 46 | |
| 47 | export async function getTask(c: AppContext, l: Loaded, number: number): Promise<TaskView> { |
| 48 | requireTaskReader(l); |
| 49 | const t = await projectStub(c.env, l.repo.id).get(number); |
| 50 | if (!t) throw new HttpError(404, "task not found"); |
| 51 | return t; |
| 52 | } |
| 53 | |
| 54 | /** The providers a run gets, checked against the runtime's needs and the person's keys. */ |
| 55 | async function providersFor(env: Env, person: Account, runtime: RuntimeId): Promise<Provider[]> { |
| 56 | const have = await new ModelKeys(env).providers(person.id); |
| 57 | for (const p of RUNTIMES[runtime].needs) { |
| 58 | if (!have.has(p)) throw new HttpError(422, `${RUNTIMES[runtime].name} needs your ${PROVIDERS[p].name} API key: add it in Settings → Model access`); |
| 59 | } |
| 60 | return runtime === "custom" ? [...have] : RUNTIMES[runtime].needs; |
| 61 | } |
| 62 | |
| 63 | async function asTaskCaller(c: AppContext, person: Account, agent: Account, l: Loaded): Promise<Caller> { |
| 64 | try { |
| 65 | return await taskCaller(c.get("dir"), person, agent, l.repo); |
| 66 | } catch (e) { |
| 67 | if (e instanceof TaskAccessError) throw new HttpError(403, e.message); |
| 68 | throw e; |
| 69 | } |
| 70 | } |
| 71 | |
| 72 | export interface StartInput { |
| 73 | goal?: unknown; |
| 74 | runtime?: unknown; |
| 75 | agent?: unknown; |
| 76 | base?: unknown; |
| 77 | command?: unknown; |
| 78 | /** Work on this pull request's branch (Fix it) instead of a new branch. */ |
| 79 | pull?: unknown; |
| 80 | } |
| 81 | |
| 82 | export async function startTask(c: AppContext, l: Loaded, input: StartInput): Promise<TaskView> { |
| 83 | const person = requirePersonSession(c); |
| 84 | if (l.repo.archived) throw new HttpError(403, "repository is archived (read-only)"); |
| 85 | const goal = typeof input.goal === "string" ? input.goal.trim() : ""; |
| 86 | if (!goal) throw new HttpError(422, "`goal` is required"); |
| 87 | if (goal.length > MAX_GOAL) throw new HttpError(422, `\`goal\` is longer than ${MAX_GOAL} characters`); |
| 88 | const runtime = typeof input.runtime === "string" ? input.runtime : "claude-code"; |
| 89 | if (!isRuntime(runtime)) throw new HttpError(422, "`runtime` must be claude-code, codex or custom"); |
| 90 | const command = runtime === "custom" ? (typeof input.command === "string" ? input.command.trim() : "") : null; |
| 91 | if (runtime === "custom" && !command) throw new HttpError(422, "`command` is required for the custom runtime"); |
| 92 | if (command && command.length > MAX_COMMAND) throw new HttpError(422, `\`command\` is longer than ${MAX_COMMAND} characters`); |
| 93 | const agentLogin = typeof input.agent === "string" ? input.agent.trim() : ""; |
| 94 | const agent = agentLogin ? await c.get("dir").accountByLogin(agentLogin) : null; |
| 95 | if (!agent || agent.kind !== "agent") throw new HttpError(422, "`agent` must be the login of one of your agents"); |
| 96 | await asTaskCaller(c, person, agent, l); |
| 97 | let base = typeof input.base === "string" && input.base.trim() ? input.base.trim() : l.repo.default_branch; |
| 98 | let pull: { number: number; head: string } | undefined; |
| 99 | if (input.pull !== undefined && input.pull !== null && input.pull !== "") { |
| 100 | const n = Number(input.pull); |
| 101 | if (!Number.isInteger(n) || n <= 0) throw new HttpError(422, "`pull` must be a pull request number"); |
| 102 | const { issue } = await l.stub.issue(n).catch(() => ({ issue: null })); |
| 103 | if (!issue?.pull) throw new HttpError(422, `#${n} isn't a pull request`); |
| 104 | if (issue.state !== "open") throw new HttpError(422, `#${n} is closed`); |
| 105 | if (!(await l.stub.getRef(`refs/heads/${issue.pull.head}`))) throw new HttpError(422, `#${n}'s branch ${issue.pull.head} isn't in this repository`); |
| 106 | pull = { number: n, head: issue.pull.head }; |
| 107 | base = issue.pull.base; |
| 108 | } |
| 109 | if (!(await l.stub.getRef(`refs/heads/${base}`))) throw new HttpError(422, `branch ${base} doesn't exist`); |
| 110 | const providers = await providersFor(c.env, person, runtime); |
| 111 | const t = await projectStub(c.env, l.repo.id).create({ repoId: l.repo.id, goal, runtime, command, agent: { id: agent.id, login: agent.login }, person: { id: person.id, login: person.login }, base, providers, pull }); |
| 112 | c.executionCtx.waitUntil(c.get("dir").audit({ actorId: person.id, actorLogin: person.login, action: "task.start", target: `${l.repo.owner_login}/${l.repo.name}#${t.number}`, detail: { agent: agent.login, runtime } })); |
| 113 | return t; |
| 114 | } |
| 115 | |
| 116 | /** The person who started the task, in a browser session, still allowed to drive its agent here. */ |
| 117 | async function starter(c: AppContext, l: Loaded, number: number): Promise<{ person: Account; task: TaskView }> { |
| 118 | const person = requirePersonSession(c); |
| 119 | const task = await getTask(c, l, number); |
| 120 | if (task.kind === "review") throw new HttpError(409, "an agent review takes no messages and isn't retried: ask for a new review on its pull request"); |
| 121 | if (task.person_id !== person.id) throw new HttpError(403, `only ${task.person_login}, who started this task, can steer or retry it: each run spends their model key`); |
| 122 | if (l.repo.archived) throw new HttpError(403, "repository is archived (read-only)"); |
| 123 | const agent = await c.get("dir").account(task.agent_id); |
| 124 | if (!agent) throw new HttpError(403, "the task's agent no longer exists"); |
| 125 | await asTaskCaller(c, person, agent, l); |
| 126 | return { person, task }; |
| 127 | } |
| 128 | |
| 129 | export async function messageTask(c: AppContext, l: Loaded, number: number, text: unknown): Promise<TaskView> { |
| 130 | const body = typeof text === "string" ? text.trim() : ""; |
| 131 | if (!body) throw new HttpError(422, "`body` is required"); |
| 132 | if (body.length > MAX_GOAL) throw new HttpError(422, `\`body\` is longer than ${MAX_GOAL} characters`); |
| 133 | const { person, task } = await starter(c, l, number); |
| 134 | const t = await projectStub(c.env, l.repo.id).message(number, body, await providersFor(c.env, person, task.runtime)); |
| 135 | c.executionCtx.waitUntil(c.get("dir").audit({ actorId: person.id, actorLogin: person.login, action: "task.message", target: `${l.repo.owner_login}/${l.repo.name}#${number}` })); |
| 136 | return t; |
| 137 | } |
| 138 | |
| 139 | export async function retryTask(c: AppContext, l: Loaded, number: number): Promise<TaskView> { |
| 140 | const { person, task } = await starter(c, l, number); |
| 141 | const t = await projectStub(c.env, l.repo.id).retry(number, await providersFor(c.env, person, task.runtime)); |
| 142 | c.executionCtx.waitUntil(c.get("dir").audit({ actorId: person.id, actorLogin: person.login, action: "task.retry", target: `${l.repo.owner_login}/${l.repo.name}#${number}` })); |
| 143 | return t; |
| 144 | } |
| 145 | |
| 146 | /** Whether `caller` can stop the task: its person, its agent, or a repository admin. */ |
| 147 | export const mayStop = (caller: Caller, t: TaskRow) => caller.id === t.person_id || caller.id === t.agent_id || atLeast(caller.role, "admin"); |
| 148 | |
| 149 | export async function stopTask(c: AppContext, l: Loaded, number: number): Promise<TaskView> { |
| 150 | const task = await getTask(c, l, number); |
| 151 | if (!mayStop(l.caller, task)) throw new HttpError(403, "only the person who started the task, its agent or a repository admin can stop it"); |
| 152 | const t = await projectStub(c.env, l.repo.id).stop(number, l.caller.login); |
| 153 | c.executionCtx.waitUntil(c.get("dir").audit({ actorId: l.caller.id, actorLogin: l.caller.login, onBehalfOf: l.caller.onBehalfOf, action: "task.stop", target: `${l.repo.owner_login}/${l.repo.name}#${number}` })); |
| 154 | return t; |
| 155 | } |
| 156 | |
| 157 | /** Agents a person can run tasks as, here: ones they manage that have write access to the repository. */ |
| 158 | export async function drivableAgents(c: AppContext, person: Account, l: Loaded): Promise<Account[]> { |
| 159 | const dir = c.get("dir"); |
| 160 | const orgs = (await dir.orgsFor(person.id)).filter((o) => o.role === "owner" || o.role === "admin").map((o) => o.id); |
| 161 | const agents = await dir.agentsOwnedBy([person.id, ...orgs]); |
| 162 | const out: Account[] = []; |
| 163 | for (const a of agents) { |
| 164 | if (!(await mayDriveAgent(dir, person, a))) continue; |
| 165 | if (atLeast(await dir.roleFor(a, l.repo), "write")) out.push(a); |
| 166 | } |
| 167 | return out; |
| 168 | } |
| 169 | |
| 170 | // ---------------------------------------------------------------- JSON |
| 171 | |
| 172 | export function taskJson(base: string, repoFull: string, t: TaskRow | TaskView) { |
| 173 | return { |
| 174 | number: t.number, |
| 175 | kind: t.kind, |
| 176 | title: t.title, |
| 177 | goal: t.goal, |
| 178 | state: t.state, |
| 179 | runtime: t.runtime, |
| 180 | ...(t.command ? { command: t.command } : {}), |
| 181 | agent: { login: t.agent_login, type: "Bot" }, |
| 182 | started_by: { login: t.person_login, type: "User" }, |
| 183 | base: t.base, |
| 184 | branch: t.branch, |
| 185 | pull_request: t.pr_number ? { number: t.pr_number, html_url: `${base}/${repoFull}/pull/${t.pr_number}` } : null, |
| 186 | queued_message: t.queued ?? null, |
| 187 | error: t.error, |
| 188 | html_url: `${base}/${repoFull}/tasks/${t.number}`, |
| 189 | created_at: t.created_at, |
| 190 | updated_at: t.updated_at, |
| 191 | ...("runs" in t ? { runs: t.runs.map((r) => ({ number: r.seq, kind: r.kind, message: r.message, state: r.step, exit_code: r.exit_code, error: r.error, model_requests: r.model_calls ?? 0, tokens: r.tokens ?? 0, started_at: r.started_at, ended_at: r.ended_at })) } : {}), |
| 192 | }; |
| 193 | } |
| 194 | |
| 195 | export const lineJson = (l: TranscriptLine) => ({ seq: l.seq, run: l.run, at: l.at, stream: l.stream, text: l.text }); |
| 196 | |
| 197 | export interface MyTask { |
| 198 | repository: string; |
| 199 | repoId: string; |
| 200 | number: number; |
| 201 | title: string; |
| 202 | runtime: RuntimeId; |
| 203 | state: TaskRow["state"]; |
| 204 | agent: string; |
| 205 | prNumber: number | null; |
| 206 | createdAt: string; |
| 207 | updatedAt: string; |
| 208 | } |
| 209 | |
| 210 | /** |
| 211 | * Tasks a person started (or, for an agent, tasks it runs), newest activity |
| 212 | * first, from the D1 index, in repositories where they still have write access. |
| 213 | */ |
| 214 | export async function tasksFor(env: Env, dir: Directory, me: Account, opts: { repoIds?: string[] | null; since?: string; activeOrSince?: boolean; limit?: number } = {}): Promise<MyTask[]> { |
| 215 | const col = me.kind === "agent" ? "t.agent_id" : "t.started_by"; |
| 216 | const since = opts.since ?? ""; |
| 217 | const r = await env.DB.prepare( |
| 218 | `SELECT t.repo_id, t.number, t.title, t.runtime, t.state, t.pr_number, t.created_at, t.updated_at, o.login AS owner_login, r.name AS repo_name, ag.login AS agent_login |
| 219 | FROM agent_tasks t JOIN repos r ON r.id = t.repo_id JOIN accounts o ON o.id = r.owner_id JOIN accounts ag ON ag.id = t.agent_id |
| 220 | WHERE ${col} = ? AND (t.updated_at >= ?${opts.activeOrSince ? " OR t.state = 'running'" : ""}) ORDER BY t.updated_at DESC LIMIT ?`, |
| 221 | ) |
| 222 | .bind(me.id, since, Math.min(opts.limit ?? 50, 100)) |
| 223 | .all<{ repo_id: string; number: number; title: string; runtime: RuntimeId; state: TaskRow["state"]; pr_number: number | null; created_at: string; updated_at: string; owner_login: string; repo_name: string; agent_login: string }>(); |
| 224 | const allowed = new Map<string, boolean>(); |
| 225 | const out: MyTask[] = []; |
| 226 | for (const t of r.results) { |
| 227 | if (opts.repoIds && !opts.repoIds.includes(t.repo_id)) continue; |
| 228 | if (!allowed.has(t.repo_id)) { |
| 229 | const repo = await dir.repoById(t.repo_id); |
| 230 | allowed.set(t.repo_id, !!repo && atLeast(await dir.roleFor(me, repo), "write")); |
| 231 | } |
| 232 | if (!allowed.get(t.repo_id)) continue; |
| 233 | out.push({ repository: `${t.owner_login}/${t.repo_name}`, repoId: t.repo_id, number: t.number, title: t.title, runtime: t.runtime, state: t.state, agent: t.agent_login, prNumber: t.pr_number, createdAt: t.created_at, updatedAt: t.updated_at }); |
| 234 | } |
| 235 | return out; |
| 236 | } |
| 237 | |
| 238 | export const myTaskJson = (base: string, t: MyTask) => ({ |
| 239 | repository: t.repository, |
| 240 | number: t.number, |
| 241 | title: t.title, |
| 242 | state: t.state, |
| 243 | runtime: t.runtime, |
| 244 | agent: { login: t.agent, type: "Bot" }, |
| 245 | pull_request: t.prNumber ? { number: t.prNumber, html_url: `${base}/${t.repository}/pull/${t.prNumber}` } : null, |
| 246 | html_url: `${base}/${t.repository}/tasks/${t.number}`, |
| 247 | created_at: t.createdAt, |
| 248 | updated_at: t.updatedAt, |
| 249 | }); |
| 250 | |
| 251 | export async function myTasks(c: AppContext, me: Account, repoIds: string[] | null) { |
| 252 | return (await tasksFor(c.env, c.get("dir"), me, { repoIds })).map((t) => myTaskJson(c.env.PUBLIC_URL.replace(/\/$/, ""), t)); |
| 253 | } |