| 1 | import { requireAccount, type AppContext } from "../auth/principal.js"; |
| 2 | import type { DiffineryEvent } from "@diffinery/shared"; |
| 3 | import type { Env } from "../env.js"; |
| 4 | import type { loadRepo } from "../http/common.js"; |
| 5 | import type { Account, Directory } from "../lib/directory.js"; |
| 6 | import { HttpError } from "../lib/util.js"; |
| 7 | import { atLeast, type Caller } from "../repo/types.js"; |
| 8 | import { mayDriveAgent, taskCaller, TaskAccessError } from "./access.js"; |
| 9 | import { isProvider, type Provider } from "./keys.js"; |
| 10 | import { payerFor } from "./org-policy.js"; |
| 11 | import type { AttemptView, TaskRow, TaskView } from "./ProjectDO.js"; |
| 12 | import { MAX_ATTEMPTS } from "./project.js"; |
| 13 | import { checksState, type ChecksState } from "./today-repo.js"; |
| 14 | import { isRuntime, MAX_COMMAND, MAX_GOAL, OPENROUTER_DEFAULT_MODEL, OPENROUTER_MODEL, RUNTIMES, runtimeLabel, runtimeNeeds, type ModelRoute, type RuntimeId, type TranscriptLine } from "./runtimes.js"; |
| 15 | import { totalUsage, usageJson } from "./usage.js"; |
| 16 | |
| 17 | /** |
| 18 | * Who may do what with agent tasks. Shared by the REST API, MCP and the web. |
| 19 | * |
| 20 | * - Reading tasks and transcripts: write access to the repository or more. |
| 21 | * - Starting, sending a message and retrying: only a person, in a browser |
| 22 | * session (each run spends their model key), and for an existing task only |
| 23 | * the person who started it. No token, agent, job or sandbox can. |
| 24 | * - Stopping: the person who started it, its agent, or a repository admin. |
| 25 | */ |
| 26 | |
| 27 | type Loaded = Awaited<ReturnType<typeof loadRepo>>; |
| 28 | |
| 29 | export const projectStub = (env: Env, repoId: string) => env.PROJECT.get(env.PROJECT.idFromName(repoId)); |
| 30 | |
| 31 | /** Tasks are visible with write access or more: transcripts show an agent's reasoning and command output. */ |
| 32 | export function requireTaskReader(l: Loaded): void { |
| 33 | if (!atLeast(l.caller.role, "write")) throw new HttpError(404, "not found"); |
| 34 | } |
| 35 | |
| 36 | /** A person signed in in a browser (or the local dev sign-in): the only one who can spend their model key. */ |
| 37 | export function requirePersonSession(c: AppContext): Account { |
| 38 | const p = c.get("principal"); |
| 39 | if (p.type === "anonymous") throw new HttpError(401, "sign in required"); |
| 40 | 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"); |
| 41 | const me = requireAccount(c); |
| 42 | if (me.kind !== "user") throw new HttpError(403, "only people can start agent tasks"); |
| 43 | return me; |
| 44 | } |
| 45 | |
| 46 | export function taskNumber(s: string | undefined): number { |
| 47 | const n = Number(s); |
| 48 | if (!Number.isInteger(n) || n <= 0) throw new HttpError(404, "task not found"); |
| 49 | return n; |
| 50 | } |
| 51 | |
| 52 | export async function getTask(c: AppContext, l: Loaded, number: number): Promise<TaskView> { |
| 53 | requireTaskReader(l); |
| 54 | const t = await projectStub(c.env, l.repo.id).get(number); |
| 55 | if (!t) throw new HttpError(404, "task not found"); |
| 56 | return t; |
| 57 | } |
| 58 | |
| 59 | /** |
| 60 | * The providers a run of `task` gets, and that whose key paid for it at its start (org-policy.ts) still may: a |
| 61 | * message, retry or plan approval keeps the task's key owner. |
| 62 | */ |
| 63 | export async function taskProviders(c: AppContext, l: Loaded, person: Account, task: TaskRow): Promise<Provider[]> { |
| 64 | return (await payerFor(c.env, c.get("dir"), l.repo, person, task.runtime, task.via, task.key_owner ?? task.person_id)).providers; |
| 65 | } |
| 66 | |
| 67 | /** |
| 68 | * Where a runtime's model calls go: `provider` is its own (the default) or |
| 69 | * OpenRouter, which takes a `model` (a default one when none is given). |
| 70 | */ |
| 71 | export function modelRoute(runtime: RuntimeId, provider: unknown, model: unknown): ModelRoute { |
| 72 | const own = RUNTIMES[runtime].needs[0]; |
| 73 | if (provider === undefined || provider === null || provider === "" || provider === own) { |
| 74 | if (model !== undefined && model !== null && model !== "") throw new HttpError(422, "`model` is only for OpenRouter: set `provider` to openrouter"); |
| 75 | return { via: null, model: null }; |
| 76 | } |
| 77 | if (typeof provider !== "string" || !isProvider(provider) || !RUNTIMES[runtime].alternatives.includes(provider)) { |
| 78 | const ok = [own, ...RUNTIMES[runtime].alternatives].filter(Boolean); |
| 79 | throw new HttpError(422, ok.length ? `\`provider\` for ${RUNTIMES[runtime].name} must be ${ok.join(" or ")}` : `${RUNTIMES[runtime].name} gets every provider you have a key for: leave \`provider\` out`); |
| 80 | } |
| 81 | const m = typeof model === "string" && model.trim() ? model.trim() : (OPENROUTER_DEFAULT_MODEL[runtime] ?? ""); |
| 82 | if (!OPENROUTER_MODEL.test(m)) throw new HttpError(422, "`model` must be an OpenRouter model id, such as anthropic/claude-sonnet-4.5"); |
| 83 | return { via: provider, model: m }; |
| 84 | } |
| 85 | |
| 86 | async function asTaskCaller(c: AppContext, person: Account, agent: Account, l: Loaded): Promise<Caller> { |
| 87 | try { |
| 88 | return await taskCaller(c.get("dir"), person, agent, l.repo); |
| 89 | } catch (e) { |
| 90 | if (e instanceof TaskAccessError) throw new HttpError(403, e.message); |
| 91 | throw e; |
| 92 | } |
| 93 | } |
| 94 | |
| 95 | export interface StartInput { |
| 96 | goal?: unknown; |
| 97 | runtime?: unknown; |
| 98 | agent?: unknown; |
| 99 | base?: unknown; |
| 100 | command?: unknown; |
| 101 | /** The runtime's own provider (the default) or `openrouter`, with `model`. */ |
| 102 | provider?: unknown; |
| 103 | model?: unknown; |
| 104 | /** Work on this pull request's branch (Fix it) instead of a new branch. */ |
| 105 | pull?: unknown; |
| 106 | /** Plan first: true or false; left out, the project's default. A project can require it. */ |
| 107 | plan_first?: unknown; |
| 108 | /** Best of N: run the first build as 1 to 4 attempts, then pick one. */ |
| 109 | attempts?: unknown; |
| 110 | /** Verify: start a fix run when the pull request's checks fail (default on, except on someone's pull request). */ |
| 111 | fix_checks?: unknown; |
| 112 | /** Land when approved: turn on auto-merge for its pull request (default off). */ |
| 113 | land?: unknown; |
| 114 | } |
| 115 | |
| 116 | export async function startTask(c: AppContext, l: Loaded, input: StartInput): Promise<TaskView> { |
| 117 | const person = requirePersonSession(c); |
| 118 | if (l.repo.archived) throw new HttpError(403, "repository is archived (read-only)"); |
| 119 | const goal = typeof input.goal === "string" ? input.goal.trim() : ""; |
| 120 | if (!goal) throw new HttpError(422, "`goal` is required"); |
| 121 | if (goal.length > MAX_GOAL) throw new HttpError(422, `\`goal\` is longer than ${MAX_GOAL} characters`); |
| 122 | const runtime = typeof input.runtime === "string" ? input.runtime : "claude-code"; |
| 123 | if (!isRuntime(runtime)) throw new HttpError(422, "`runtime` must be claude-code, codex or custom"); |
| 124 | const command = runtime === "custom" ? (typeof input.command === "string" ? input.command.trim() : "") : null; |
| 125 | if (runtime === "custom" && !command) throw new HttpError(422, "`command` is required for the custom runtime"); |
| 126 | if (command && command.length > MAX_COMMAND) throw new HttpError(422, `\`command\` is longer than ${MAX_COMMAND} characters`); |
| 127 | const route = modelRoute(runtime, input.provider, input.model); |
| 128 | const agentLogin = typeof input.agent === "string" ? input.agent.trim() : ""; |
| 129 | const agent = agentLogin ? await c.get("dir").accountByLogin(agentLogin) : null; |
| 130 | if (!agent || agent.kind !== "agent") throw new HttpError(422, "`agent` must be the login of one of your agents"); |
| 131 | await asTaskCaller(c, person, agent, l); |
| 132 | let base = typeof input.base === "string" && input.base.trim() ? input.base.trim() : l.repo.default_branch; |
| 133 | let pull: { number: number; head: string } | undefined; |
| 134 | if (input.pull !== undefined && input.pull !== null && input.pull !== "") { |
| 135 | const n = Number(input.pull); |
| 136 | if (!Number.isInteger(n) || n <= 0) throw new HttpError(422, "`pull` must be a pull request number"); |
| 137 | const { issue } = await l.stub.issue(n).catch(() => ({ issue: null })); |
| 138 | if (!issue?.pull) throw new HttpError(422, `#${n} isn't a pull request`); |
| 139 | if (issue.state !== "open") throw new HttpError(422, `#${n} is closed`); |
| 140 | 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`); |
| 141 | pull = { number: n, head: issue.pull.head }; |
| 142 | base = issue.pull.base; |
| 143 | } |
| 144 | if (!(await l.stub.getRef(`refs/heads/${base}`))) throw new HttpError(422, `branch ${base} doesn't exist`); |
| 145 | if (input.plan_first !== undefined && typeof input.plan_first !== "boolean") throw new HttpError(422, "`plan_first` must be true or false"); |
| 146 | const attempts = input.attempts === undefined || input.attempts === null || input.attempts === "" ? 1 : Number(input.attempts); |
| 147 | if (!Number.isInteger(attempts) || attempts < 1 || attempts > MAX_ATTEMPTS) throw new HttpError(422, `\`attempts\` must be a whole number from 1 to ${MAX_ATTEMPTS}`); |
| 148 | if (attempts > 1 && pull) throw new HttpError(422, "a task on a pull request's branch runs one attempt"); |
| 149 | for (const k of ["fix_checks", "land"] as const) if (input[k] !== undefined && typeof input[k] !== "boolean") throw new HttpError(422, `\`${k}\` must be true or false`); |
| 150 | const fixChecks = input.fix_checks === undefined ? !pull : input.fix_checks === true; |
| 151 | const land = input.land === true; |
| 152 | const { providers, keyOwner, orgMaxRunning } = await payerFor(c.env, c.get("dir"), l.repo, person, runtime, route.via); |
| 153 | const project = projectStub(c.env, l.repo.id); |
| 154 | const { planFirst: policy } = await project.projectSettings(); |
| 155 | const planFirst = policy === "required" || (input.plan_first === undefined ? policy === "default" : input.plan_first === true); |
| 156 | const t = await project.create({ repoId: l.repo.id, goal, runtime, command, route, agent: { id: agent.id, login: agent.login }, person: { id: person.id, login: person.login }, base, providers, pull, planFirst, attempts, fixChecks, land, keyOwner, orgMaxRunning }); |
| 157 | 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, ...(keyOwner !== person.id ? { paid_by: l.repo.owner_login } : {}), ...(route.via ? { provider: route.via, model: route.model } : {}), ...(planFirst ? { plan_first: true } : {}), ...(attempts > 1 ? { attempts } : {}), fix_checks: fixChecks, land } })); |
| 158 | return t; |
| 159 | } |
| 160 | |
| 161 | /** The person who started the task, in a browser session, still allowed to drive its agent here. */ |
| 162 | export async function starter(c: AppContext, l: Loaded, number: number): Promise<{ person: Account; task: TaskView }> { |
| 163 | const person = requirePersonSession(c); |
| 164 | const task = await getTask(c, l, number); |
| 165 | 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"); |
| 166 | if (task.kind === "coordinator") throw new HttpError(409, "a coordinator takes no messages and isn't retried: post in the project instead"); |
| 167 | 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`); |
| 168 | if (l.repo.archived) throw new HttpError(403, "repository is archived (read-only)"); |
| 169 | const agent = await c.get("dir").account(task.agent_id); |
| 170 | if (!agent) throw new HttpError(403, "the task's agent no longer exists"); |
| 171 | await asTaskCaller(c, person, agent, l); |
| 172 | return { person, task }; |
| 173 | } |
| 174 | |
| 175 | export async function messageTask(c: AppContext, l: Loaded, number: number, text: unknown): Promise<TaskView> { |
| 176 | const body = typeof text === "string" ? text.trim() : ""; |
| 177 | if (!body) throw new HttpError(422, "`body` is required"); |
| 178 | if (body.length > MAX_GOAL) throw new HttpError(422, `\`body\` is longer than ${MAX_GOAL} characters`); |
| 179 | const { person, task } = await starter(c, l, number); |
| 180 | const t = await projectStub(c.env, l.repo.id).message(number, body, await taskProviders(c, l, person, task)); |
| 181 | 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}` })); |
| 182 | return t; |
| 183 | } |
| 184 | |
| 185 | /** Verify and land: change a task's options (the person who started it, in a browser). */ |
| 186 | export async function setTaskOptions(c: AppContext, l: Loaded, number: number, input: { fix_checks?: unknown; land?: unknown }): Promise<TaskView> { |
| 187 | const { person } = await starter(c, l, number); |
| 188 | const o: { fixChecks?: boolean; land?: boolean } = {}; |
| 189 | for (const [k, to] of [["fix_checks", "fixChecks"], ["land", "land"]] as const) { |
| 190 | const v = input[k]; |
| 191 | if (v === undefined) continue; |
| 192 | if (typeof v !== "boolean") throw new HttpError(422, `\`${k}\` must be true or false`); |
| 193 | o[to] = v; |
| 194 | } |
| 195 | const t = await projectStub(c.env, l.repo.id).setOptions(number, o, person.login); |
| 196 | c.executionCtx.waitUntil(c.get("dir").audit({ actorId: person.id, actorLogin: person.login, action: "task.options", target: `${l.repo.owner_login}/${l.repo.name}#${number}`, detail: { fix_checks: !!t.fix_checks, land: !!t.land } })); |
| 197 | return t; |
| 198 | } |
| 199 | |
| 200 | /** |
| 201 | * Verify (phase 3e): a workflow run or a commit status finished. If the repository has agent tasks with a pull |
| 202 | * request, its project checks whether any of them should start a fix run. Called from the event queue, best effort. |
| 203 | */ |
| 204 | export async function onVerifyEvent(env: Env, ev: DiffineryEvent): Promise<void> { |
| 205 | if (!((ev.type === "workflow_run" && ev.action === "completed") || ev.type === "status")) return; |
| 206 | const any = await env.DB.prepare("SELECT 1 FROM agent_tasks WHERE repo_id = ? AND state = 'ready' AND pr_number IS NOT NULL LIMIT 1").bind(ev.repoId).first(); |
| 207 | if (!any) return; |
| 208 | await projectStub(env, ev.repoId).verify(); |
| 209 | } |
| 210 | |
| 211 | /** Best of N: pick one of a task's attempts, which becomes its branch and pull request. */ |
| 212 | export async function pickAttempt(c: AppContext, l: Loaded, number: number, attempt: number): Promise<TaskView> { |
| 213 | const { person } = await starter(c, l, number); |
| 214 | const t = await projectStub(c.env, l.repo.id).pick(number, attempt, person.login); |
| 215 | c.executionCtx.waitUntil(c.get("dir").audit({ actorId: person.id, actorLogin: person.login, action: "task.pick", target: `${l.repo.owner_login}/${l.repo.name}#${number}`, detail: { attempt } })); |
| 216 | return t; |
| 217 | } |
| 218 | |
| 219 | /** One attempt as the comparison shows it: what it changed, its checks, what it cost, what it said. */ |
| 220 | export interface AttemptCompare extends AttemptView { |
| 221 | files: number; |
| 222 | additions: number; |
| 223 | deletions: number; |
| 224 | checks: { state: ChecksState; failing: string[] }; |
| 225 | } |
| 226 | |
| 227 | export async function compareAttempts(c: AppContext, l: Loaded, number: number): Promise<AttemptCompare[]> { |
| 228 | const list = await projectStub(c.env, l.repo.id).attempts(number); |
| 229 | return Promise.all( |
| 230 | list.map(async (a) => { |
| 231 | if (!a.head_sha) return { ...a, files: 0, additions: 0, deletions: 0, checks: { state: "none" as ChecksState, failing: [] } }; |
| 232 | const [d, checks] = await Promise.all([l.stub.diff(a.start_sha, a.head_sha, { patch: false }).catch(() => null), l.stub.checksAndStatuses(a.head_sha).catch(() => [])]); |
| 233 | return { ...a, files: d?.totalFiles ?? 0, additions: d?.additions ?? 0, deletions: d?.deletions ?? 0, checks: checksState(checks) }; |
| 234 | }), |
| 235 | ); |
| 236 | } |
| 237 | |
| 238 | export const attemptJson = (base: string, repoFull: string, a: AttemptCompare) => ({ |
| 239 | attempt: a.attempt, |
| 240 | run: a.run, |
| 241 | state: a.step, |
| 242 | error: a.error, |
| 243 | branch: a.branch, |
| 244 | head_sha: a.head_sha, |
| 245 | changes: { files: a.files, additions: a.additions, deletions: a.deletions }, |
| 246 | checks: a.checks.state, |
| 247 | failing_checks: a.checks.failing, |
| 248 | model_requests: a.model_calls, |
| 249 | tokens: a.tokens, |
| 250 | summary: a.summary, |
| 251 | ...(a.head_sha ? { compare_url: `${base}/${repoFull}/compare/${a.start_sha ?? ""}...${a.branch}` } : {}), |
| 252 | }); |
| 253 | |
| 254 | /** Approve a task's waiting plan: its build run starts. */ |
| 255 | export async function approvePlan(c: AppContext, l: Loaded, number: number): Promise<TaskView> { |
| 256 | const { person, task } = await starter(c, l, number); |
| 257 | const t = await projectStub(c.env, l.repo.id).approvePlan(number, await taskProviders(c, l, person, task), person.login); |
| 258 | c.executionCtx.waitUntil(c.get("dir").audit({ actorId: person.id, actorLogin: person.login, action: "task.plan_approve", target: `${l.repo.owner_login}/${l.repo.name}#${number}` })); |
| 259 | return t; |
| 260 | } |
| 261 | |
| 262 | export async function retryTask(c: AppContext, l: Loaded, number: number): Promise<TaskView> { |
| 263 | const { person, task } = await starter(c, l, number); |
| 264 | const t = await projectStub(c.env, l.repo.id).retry(number, await taskProviders(c, l, person, task)); |
| 265 | 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}` })); |
| 266 | return t; |
| 267 | } |
| 268 | |
| 269 | /** Whether `caller` can stop the task: its person, its agent, or a repository admin. */ |
| 270 | export const mayStop = (caller: Caller, t: TaskRow) => caller.id === t.person_id || caller.id === t.agent_id || atLeast(caller.role, "admin"); |
| 271 | |
| 272 | export async function stopTask(c: AppContext, l: Loaded, number: number): Promise<TaskView> { |
| 273 | const task = await getTask(c, l, number); |
| 274 | 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"); |
| 275 | const t = await projectStub(c.env, l.repo.id).stop(number, l.caller.login); |
| 276 | 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}` })); |
| 277 | return t; |
| 278 | } |
| 279 | |
| 280 | /** Agents a person can run tasks as, here: ones they manage that have write access to the repository. */ |
| 281 | export async function drivableAgents(c: AppContext, person: Account, l: Loaded): Promise<Account[]> { |
| 282 | const dir = c.get("dir"); |
| 283 | const orgs = (await dir.orgsFor(person.id)).filter((o) => o.role === "owner" || o.role === "admin").map((o) => o.id); |
| 284 | const agents = await dir.agentsOwnedBy([person.id, ...orgs]); |
| 285 | const out: Account[] = []; |
| 286 | for (const a of agents) { |
| 287 | if (!(await mayDriveAgent(dir, person, a))) continue; |
| 288 | if (atLeast(await dir.roleFor(a, l.repo), "write")) out.push(a); |
| 289 | } |
| 290 | return out; |
| 291 | } |
| 292 | |
| 293 | // ---------------------------------------------------------------- JSON |
| 294 | |
| 295 | export function taskJson(base: string, repoFull: string, t: TaskRow | TaskView) { |
| 296 | return { |
| 297 | number: t.number, |
| 298 | kind: t.kind, |
| 299 | title: t.title, |
| 300 | goal: t.goal, |
| 301 | state: t.state, |
| 302 | runtime: t.runtime, |
| 303 | ...(t.command ? { command: t.command } : {}), |
| 304 | provider: t.via ?? RUNTIMES[t.runtime].needs[0] ?? null, |
| 305 | model: t.model ?? null, |
| 306 | agent: { login: t.agent_login, type: "Bot" }, |
| 307 | started_by: { login: t.person_login, type: "User" }, |
| 308 | // Phase 5b: whose key pays, fixed when the task started: the person's own, or the organization's. |
| 309 | paid_by: t.key_owner && t.key_owner !== t.person_id ? "organization" : "person", |
| 310 | plan_first: !!t.plan_first, |
| 311 | ...(t.plan_first ? { plan_state: t.plan_state, plan: t.plan } : {}), |
| 312 | ...(t.parent_task ? { parent_task: t.parent_task } : {}), |
| 313 | attempts: t.attempts ?? 1, |
| 314 | ...((t.attempts ?? 1) > 1 ? { pick_state: t.pick_state, picked_attempt: t.picked } : {}), |
| 315 | fix_checks: !!t.fix_checks, |
| 316 | check_fixes: t.check_fixes ?? 0, |
| 317 | land: !!t.land, |
| 318 | // Security 3b: a task a coordinator started says so, with the post it came from. |
| 319 | ...(t.coordinator_post ? { started_via: { type: "coordinator", post: t.coordinator_post }, token_budget: t.token_budget, waits_for: t.after_tasks ? (JSON.parse(t.after_tasks) as number[]) : [] } : {}), |
| 320 | base: t.base, |
| 321 | branch: t.branch, |
| 322 | pull_request: t.pr_number ? { number: t.pr_number, html_url: `${base}/${repoFull}/pull/${t.pr_number}` } : null, |
| 323 | queued_message: t.queued ?? null, |
| 324 | error: t.error, |
| 325 | html_url: `${base}/${repoFull}/tasks/${t.number}`, |
| 326 | created_at: t.created_at, |
| 327 | updated_at: t.updated_at, |
| 328 | ...("runs" in t ? { usage: usageOrNull(totalUsage(t.runs.map((r) => r.usage))), 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, usage: usageOrNull(r.usage), ...(r.attempt ? { attempt: r.attempt } : {}), ...(r.terminal_id ? { in_terminal: true } : {}), started_at: r.started_at, ended_at: r.ended_at })) } : {}), |
| 329 | }; |
| 330 | } |
| 331 | |
| 332 | const usageOrNull = (u: Parameters<typeof usageJson>[0] | null) => (u ? usageJson(u) : null); |
| 333 | |
| 334 | export const lineJson = (l: TranscriptLine) => ({ seq: l.seq, run: l.run, at: l.at, stream: l.stream, text: l.text }); |
| 335 | |
| 336 | export interface MyTask { |
| 337 | repository: string; |
| 338 | repoId: string; |
| 339 | number: number; |
| 340 | title: string; |
| 341 | runtime: RuntimeId; |
| 342 | state: TaskRow["state"]; |
| 343 | agent: string; |
| 344 | prNumber: number | null; |
| 345 | createdAt: string; |
| 346 | updatedAt: string; |
| 347 | } |
| 348 | |
| 349 | /** |
| 350 | * Tasks a person started (or, for an agent, tasks it runs), newest activity |
| 351 | * first, from the D1 index, in repositories where they still have write access. |
| 352 | */ |
| 353 | export async function tasksFor(env: Env, dir: Directory, me: Account, opts: { repoIds?: string[] | null; since?: string; activeOrSince?: boolean; limit?: number } = {}): Promise<MyTask[]> { |
| 354 | const col = me.kind === "agent" ? "t.agent_id" : "t.started_by"; |
| 355 | const since = opts.since ?? ""; |
| 356 | const r = await env.DB.prepare( |
| 357 | `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 |
| 358 | 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 |
| 359 | WHERE ${col} = ? AND (t.updated_at >= ?${opts.activeOrSince ? " OR t.state IN ('running', 'queued')" : ""}) ORDER BY t.updated_at DESC LIMIT ?`, |
| 360 | ) |
| 361 | .bind(me.id, since, Math.min(opts.limit ?? 50, 100)) |
| 362 | .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 }>(); |
| 363 | const allowed = new Map<string, boolean>(); |
| 364 | const out: MyTask[] = []; |
| 365 | for (const t of r.results) { |
| 366 | if (opts.repoIds && !opts.repoIds.includes(t.repo_id)) continue; |
| 367 | if (!allowed.has(t.repo_id)) { |
| 368 | const repo = await dir.repoById(t.repo_id); |
| 369 | allowed.set(t.repo_id, !!repo && atLeast(await dir.roleFor(me, repo), "write")); |
| 370 | } |
| 371 | if (!allowed.get(t.repo_id)) continue; |
| 372 | 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 }); |
| 373 | } |
| 374 | return out; |
| 375 | } |
| 376 | |
| 377 | export const myTaskJson = (base: string, t: MyTask) => ({ |
| 378 | repository: t.repository, |
| 379 | number: t.number, |
| 380 | title: t.title, |
| 381 | state: t.state, |
| 382 | runtime: t.runtime, |
| 383 | agent: { login: t.agent, type: "Bot" }, |
| 384 | pull_request: t.prNumber ? { number: t.prNumber, html_url: `${base}/${t.repository}/pull/${t.prNumber}` } : null, |
| 385 | html_url: `${base}/${t.repository}/tasks/${t.number}`, |
| 386 | created_at: t.createdAt, |
| 387 | updated_at: t.updatedAt, |
| 388 | }); |
| 389 | |
| 390 | export async function myTasks(c: AppContext, me: Account, repoIds: string[] | null) { |
| 391 | return (await tasksFor(c.env, c.get("dir"), me, { repoIds })).map((t) => myTaskJson(c.env.PUBLIC_URL.replace(/\/$/, ""), t)); |
| 392 | } |