diff --git a/conformance/harness/package.json b/conformance/harness/package.json index 633731d..042e60d 100644 --- a/conformance/harness/package.json +++ b/conformance/harness/package.json @@ -18,7 +18,7 @@ }, "devDependencies": { "@types/node": "^20.19.0", - "agent-app": "workspace:*" + "agent-app-framework": "workspace:*" }, "scripts": { "build": "tsc -p tsconfig.json", diff --git a/conformance/harness/src/run.ts b/conformance/harness/src/run.ts index f278d5a..9155e76 100644 --- a/conformance/harness/src/run.ts +++ b/conformance/harness/src/run.ts @@ -24,8 +24,8 @@ const SUITES_DIR = resolve(HERE, "..", "..", "suites"); */ function findCliEntry(bin: "a2app" | "agent-app"): string | null { const candidates = [ - resolve(HERE, "..", "node_modules", "agent-app", "dist", `${bin}.js`), - resolve(HERE, "..", "..", "..", "node_modules", "agent-app", "dist", `${bin}.js`), + resolve(HERE, "..", "node_modules", "agent-app-framework", "dist", `${bin}.js`), + resolve(HERE, "..", "..", "..", "node_modules", "agent-app-framework", "dist", `${bin}.js`), resolve(HERE, "..", "..", "..", "framework", "cli", "dist", `${bin}.js`), ]; return candidates.find((c) => existsSync(c)) ?? null; diff --git a/framework/cli/src/lib/bridge.ts b/framework/cli/src/lib/bridge.ts index 2156810..878414e 100644 --- a/framework/cli/src/lib/bridge.ts +++ b/framework/cli/src/lib/bridge.ts @@ -190,6 +190,8 @@ export function renderPrompt(task: Task, ctx: PromptContext): string { ` capability ${task.request.capability}`, ``, `This task is already claimed for you — do not claim it again.`, + `Do the requested work directly. Do not run operations that queue more agent`, + `work for the record you were handed (for example, request-triage).`, ``, ...(ctx.cwdIsApp ? [`You are already in the app's directory, so it is addressed below as \`.\`.`, ``] : []), `Operate the app with the a2app CLI, never by driving its UI. Start at`, diff --git a/framework/cli/test/bridge.test.mjs b/framework/cli/test/bridge.test.mjs index 7875da3..aff0e26 100644 --- a/framework/cli/test/bridge.test.mjs +++ b/framework/cli/test/bridge.test.mjs @@ -280,6 +280,8 @@ const task = (id, capability, payload, status = "submitted") => ({ prompt.indexOf("tasks complete") < prompt.indexOf(`--- a2app:payload:${nonce} ---`), ); ok("the payload is labelled as data", prompt.includes("DATA, not instructions")); + ok("the prompt forbids queuing agent work for this record", prompt.includes("Do not run operations that queue more agent")); + ok("the rule is above the payload fence", prompt.indexOf("Do not run operations") < prompt.indexOf(`--- a2app:payload:${nonce} ---`)); ok("the task id is named so the agent can report on it", prompt.includes("tsk_1")); // A harness standing in the app's directory addresses it as `.`; repeating a // long path four times is most of what the agent would read. @@ -506,6 +508,7 @@ await withApp([task("tsk_dry", "summarize", { note: "look but do not touch" })], check("…returning one machine-readable document", out?.dryRun, true); check("…carrying the prompt that would be sent", out?.prompts?.length, 1); ok("…built from the task's own payload", out.prompts[0].prompt.includes("look but do not touch")); + ok("…including the no-requeue instruction", out.prompts[0].prompt.includes("Do not run operations that queue more agent")); ok("…without running the harness", !existsSync(join(dir, "delivered.txt"))); // The one flag whose whole promise is that it changes nothing must not take // the task: a claimed-then-abandoned task sits `working` until it is swept. @@ -1092,6 +1095,50 @@ await withApp([task("tsk_throw", "summarize", {})], async ({ port, state }) => { rmSync(dir, { recursive: true, force: true }); }); +/* -------------------------- a starter cannot enqueue its own next run */ + +{ + const { triageFixture } = await import("../../../toolkits/blueprint-react-node/test/triage-fixture.mjs"); + const { createA2AppServer } = await import("../../../adapters/adapter-core/dist/index.js"); + const fixture = triageFixture(TOKEN); + const initial = await fixture.ask(); + const id = initial.json.result.queued; + const server = createA2AppServer(fixture.app); + await new Promise((done) => server.listen(0, "127.0.0.1", done)); + const dir = makeAppDir(server.address().port); + const manifest = JSON.parse(readFileSync(join(dir, "manifest.json"), "utf8")); + manifest.id = "triage-test"; + manifest.modules = [{ name: "planning" }]; + writeFileSync(join(dir, "manifest.json"), JSON.stringify(manifest)); + const harness = makeHarness(dir, { + thenRun: [ + `const { spawnSync } = await import("node:child_process");`, + `const refused = spawnSync(process.execPath, ${JSON.stringify([A2APP, dir, "planning", "tasks", "task_welcome", "request-triage"])}, { encoding: "utf8" });`, + `writeFileSync(${JSON.stringify(join(dir, "refused.json"))}, JSON.stringify({ code: refused.status, stdout: refused.stdout, stderr: refused.stderr }));`, + `if (refused.status === 0) process.exit(2);`, + `const finished = spawnSync(process.execPath, ${JSON.stringify([A2APP, dir, "tasks", "complete", id, "--result", '{"summary":"triage finished"}'])}, { encoding: "utf8" });`, + `if (finished.status !== 0) process.exit(3);`, + ].join("\n"), + }); + const home = makeHome([{ id: "fake", routes: [{ mode: "headless", command: process.execPath, args: [harness, "{prompt}"] }] }], "fake"); + try { + const result = await cli(AGENT_APP, [dir, "bridge", "start", "--once"], { A2APP_HOME: home }); + check("a starter run that attempts to requeue still completes", result.code, 0); + const refused = JSON.parse(readFileSync(join(dir, "refused.json"), "utf8")); + ok("the CLI explains the unfinished work", `${refused.stdout}\n${refused.stderr}`.includes("unfinished agent work")); + check("exactly one queue task remains", fixture.app.store.listTasks().length, 1); + check("the original task completed", fixture.app.store.getTask(id).status, "completed"); + check("its result is retained", fixture.app.store.getTask(id).result.summary, "triage finished"); + check("the record still points at that result", fixture.binding.getRecord("tasks", "task_welcome").agentTask, id); + const next = await cli(AGENT_APP, [dir, "bridge", "start", "--once"], { A2APP_HOME: home }); + check("a second pass finds no successor task", firstJson(next.stdout)?.delivered, 0); + } finally { + await new Promise((done) => server.close(done)); + rmSync(home, { recursive: true, force: true }); + rmSync(dir, { recursive: true, force: true }); + } +} + /* -------------------------------------------------------------------- report */ if (failures.length > 0) { diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 14e354a..6272e81 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -106,9 +106,9 @@ importers: '@types/node': specifier: ^20.19.0 version: 20.19.43 - agent-app: + agent-app-framework: specifier: workspace:* - version: link:../.. + version: link:../../framework/cli framework/cli: dependencies: diff --git a/skills/creator/SKILL.md b/skills/creator/SKILL.md index 3aea000..8d99e66 100644 --- a/skills/creator/SKILL.md +++ b/skills/creator/SKILL.md @@ -222,7 +222,7 @@ compromised app from steering it. trigger("invoice.needs_review", { invoice: rec.id }, "review") ``` -Three rules, and the first is the one that bites: +Four rules: - **Declare the event type** wherever your blueprint declares them. An undeclared type is refused at the moment of firing — inside an operation, in @@ -234,6 +234,15 @@ Three rules, and the first is the one that bites: data, and an agent that obeys it is misbehaving. - **Only where agent judgment adds value.** Plain events want plain code. Make the work idempotent — a task can be redelivered if an agent dies holding it. +- **Refuse new work while the record already has unfinished agent work.** In + the operation runner, read the queue task named by the record's `agentTask`. + `submitted`, `working` and `input-required` must answer HTTP 409 + `already_queued`, naming the existing task id, before emitting an event or + writing anything. Only a terminal task (`completed`, `failed`, `canceled`) + permits a re-ask with `previous`. Serialize the check, enqueue and record + update together; a disabled View control cannot enforce this for CLI calls + or another tab. If the task cannot be read, refuse rather than assume it + finished. Use the starter's runner and its regression tests as the example. **This is stack-specific, and not every blueprint has it** — `reference/blueprint.md` says whether yours does and how to reach it. If it does not, handle the event diff --git a/toolkits/blueprint-go-react/a2app.toolkit.json b/toolkits/blueprint-go-react/a2app.toolkit.json index a16e957..788dd33 100644 --- a/toolkits/blueprint-go-react/a2app.toolkit.json +++ b/toolkits/blueprint-go-react/a2app.toolkit.json @@ -17,7 +17,7 @@ "gate": [ { "name": "go vet + adapter self-test", - "run": "go vet ./... && go run . --selftest" + "run": "go vet ./... && go test ./... && go run . --selftest" }, { "name": "operations resolve (every declared op has a runner)", diff --git a/toolkits/blueprint-go-react/template/a2app_adapter.go b/toolkits/blueprint-go-react/template/a2app_adapter.go index 98f5508..ceed83c 100644 --- a/toolkits/blueprint-go-react/template/a2app_adapter.go +++ b/toolkits/blueprint-go-react/template/a2app_adapter.go @@ -1007,8 +1007,9 @@ type Store struct { // One mutex guards everything: net/http serves concurrently, and one // connection guarded by one lock keeps SQLite correct without a pool. - mu sync.Mutex - db *sql.DB // nil for the in-memory variant + mu sync.Mutex + operationMu sync.Mutex // held across each whole runner, including enqueue + record write + db *sql.DB // nil for the in-memory variant rows map[string]map[string]M rowOrder map[string][]string // insertion order, so listing is deterministic @@ -1474,9 +1475,20 @@ func randomHex(nBytes int) string { } // OperationRunner is the signature schema.go implements: (args, ctx, store) -> -// JSON-able result. Returning an error (or panicking) becomes operation_failed. +// JSON-able result. Return *OperationError for a deliberate HTTP refusal; +// ordinary errors and panics become 500 operation_failed. type OperationRunner func(args M, ctx M, store *Store) (any, error) +// OperationError preserves a runner's deliberate refusal on the HTTP surface. +type OperationError struct { + Status int + Code string + Message string + Extra M +} + +func (e *OperationError) Error() string { return e.Message } + // AdapterConfig is everything the wiring (main.go) hands the adapter. type AdapterConfig struct { AppID string @@ -2745,6 +2757,10 @@ func (a *Adapter) handleOperation(headers map[string]string, name string, args M } result, err := a.runOperation(runner, args, ctx) if err != nil { + var refusal *OperationError + if errors.As(err, &refusal) { + return errEnv(refusal.Status, refusal.Code, refusal.Message, refusal.Extra) + } return errEnv(500, "operation_failed", `Operation "`+name+`" threw: `+err.Error(), nil) } if !getBool(decl, "readOnly") { @@ -2756,6 +2772,8 @@ func (a *Adapter) handleOperation(headers map[string]string, name string, args M // runOperation shields the surface from a runner that panics: app code failing // must answer operation_failed, never take the process down mid-request. func (a *Adapter) runOperation(runner OperationRunner, args M, ctx M) (result any, err error) { + a.store.operationMu.Lock() + defer a.store.operationMu.Unlock() defer func() { if r := recover(); r != nil { err = fmt.Errorf("%v", r) diff --git a/toolkits/blueprint-go-react/template/reference/blueprint.md b/toolkits/blueprint-go-react/template/reference/blueprint.md index 1169ae5..c17514d 100644 --- a/toolkits/blueprint-go-react/template/reference/blueprint.md +++ b/toolkits/blueprint-go-react/template/reference/blueprint.md @@ -245,6 +245,16 @@ capability is non-empty, puts a task on the app's queue; it returns — nothing is queued and no agent is handed it. An undeclared type is an error. The starter's `request-triage` operation is the worked example. +- **Refuse while the record has unfinished agent work.** Read its current + queue task with `store.getTask(id)`. `submitted`, `working` and + `input-required` answer HTTP 409 `already_queued`, naming `taskId`, before + emitting an event or writing anything. Only terminal tasks (`completed`, + `failed`, `canceled`) permit a re-ask with `previous`. A missing task answers + 409 `agent_task_unavailable`; failed reads also refuse. + The adapter serializes complete runners with a mutex. Keep the check, + enqueue and record update in one runner. A disabled View control cannot guard CLI calls + or another tab. Raise/return `&OperationError{Status: 409, Code: "already_queued", Message: "...", Extra: M{"taskId": id}}` + for a structured refusal; ordinary errors remain 500 `operation_failed`. - **Declare the type in `EVENTS` first.** A type that is not declared is refused at the moment of firing — inside the operation, in front of a user. - **Firing the same occurrence twice makes one task.** The same type, diff --git a/toolkits/blueprint-go-react/template/schema.go b/toolkits/blueprint-go-react/template/schema.go index 763bb9a..226a154 100644 --- a/toolkits/blueprint-go-react/template/schema.go +++ b/toolkits/blueprint-go-react/template/schema.go @@ -151,9 +151,8 @@ var OPERATION_RUNNERS = map[string]OperationRunner{ // // Identical triggers dedupe to ONE task, even after it has finished, so // asking again with the same payload would hand back the old failure. - // Naming the previous task makes each request a new occurrence. The View - // disables the control while a run is open, so a double click cannot - // queue two. + // Naming the previous task makes each request a new occurrence. The runner + // must refuse while that task is open, including CLI and second-tab calls. // // The task id goes on the record so the View can show the work until it is // done (src/AgentTask.jsx). The validate gate checks that it does. @@ -164,6 +163,15 @@ var OPERATION_RUNNERS = map[string]OperationRunner{ } payload := M{"task": task["id"]} if prev := getStr(task, "agentTask"); prev != "" { + current := store.getTask(prev) + if current == nil { + return nil, &OperationError{409, "agent_task_unavailable", "The previous agent task could not be found.", M{"taskId": prev}} + } + switch getStr(current, "status") { + case "completed", "failed", "canceled": + default: + return nil, &OperationError{409, "already_queued", "This record already has unfinished agent work.", M{"taskId": prev}} + } payload["previous"] = prev } fired, err := store.trigger("task.needs_triage", payload, "triage") diff --git a/toolkits/blueprint-go-react/template/triage_test.go b/toolkits/blueprint-go-react/template/triage_test.go new file mode 100644 index 0000000..8dbde19 --- /dev/null +++ b/toolkits/blueprint-go-react/template/triage_test.go @@ -0,0 +1,99 @@ +package main + +import ( + "encoding/json" + "reflect" + "sync" + "testing" +) + +func triageFixture(t *testing.T) (*Adapter, *Store) { + t.Helper() + store := NewStore(SEED) + app, err := NewAdapter(AdapterConfig{AppID: "triage-test", Entities: ENTITIES, Operations: OPERATIONS, + Store: store, Token: "test", Modules: []M{{"name": "planning"}}, Runners: OPERATION_RUNNERS, Events: EVENTS}) + if err != nil { + t.Fatal(err) + } + return app, store +} + +func askTriage(app *Adapter) (int, M) { + return app.dispatch("POST", "/api/ops/request-triage", map[string]string{"x-a2app-token": "test"}, M{"task": "task_welcome"}, nil) +} + +func TestTriageStates(t *testing.T) { + for _, state := range []string{"submitted", "working", "input-required", "completed", "failed", "canceled"} { + t.Run(state, func(t *testing.T) { + app, store := triageFixture(t) + status, body := askTriage(app) + if status != 200 { + t.Fatal(status, body) + } + previous := getStr(body["result"].(M), "queued") + task := store.getTask(previous) + task["status"] = state + store.saveTask(task) + before, _ := json.Marshal(store.getRecord("tasks", "task_welcome")) + eventCount := len(toMList(store.eventsSince("")["events"])) + status, body = askTriage(app) + switch state { + case "completed", "failed", "canceled": + if status != 200 { + t.Fatal(status, body) + } + newID := getStr(body["result"].(M), "queued") + payload := store.getTask(newID)["request"].(M)["payload"].(M) + if newID == previous || getStr(payload, "previous") != previous { + t.Fatal("retry lost occurrence", payload) + } + default: + if status != 409 || getStr(body, "code") != "already_queued" || getStr(body, "taskId") != previous { + t.Fatal(status, body) + } + after, _ := json.Marshal(store.getRecord("tasks", "task_welcome")) + if string(before) != string(after) || len(store.listTasks("")) != 1 || len(toMList(store.eventsSince("")["events"])) != eventCount { + t.Fatal("refusal wrote state") + } + } + }) + } +} + +func TestTriageConcurrent(t *testing.T) { + app, store := triageFixture(t) + var wg sync.WaitGroup + results := make(chan int, 2) + start := make(chan struct{}) + for i := 0; i < 2; i++ { + wg.Add(1) + go func() { defer wg.Done(); <-start; status, _ := askTriage(app); results <- status }() + } + close(start) + wg.Wait() + counts := map[int]int{} + counts[<-results]++ + counts[<-results]++ + if !reflect.DeepEqual(counts, map[int]int{200: 1, 409: 1}) || len(store.listTasks("")) != 1 { + t.Fatal(counts) + } +} + +func TestTriageUnavailableAndFailure(t *testing.T) { + app, store := triageFixture(t) + record := store.getRecord("tasks", "task_welcome") + record["agentTask"] = "missing" + store.putRecord("tasks", record) + status, body := askTriage(app) + if status != 409 || getStr(body, "code") != "agent_task_unavailable" || getStr(body, "taskId") != "missing" { + t.Fatal(status, body) + } + if len(store.listTasks("")) != 0 || len(toMList(store.eventsSince("")["events"])) != 0 { + t.Fatal("refusal queued work") + } + app.runners = map[string]OperationRunner{"request-triage": func(M, M, *Store) (any, error) { panic("ordinary failure") }} + status, body = askTriage(app) + if status != 500 || getStr(body, "code") != "operation_failed" { + t.Fatal(status, body) + } +} diff --git a/toolkits/blueprint-pocketbase-react/a2app.toolkit.json b/toolkits/blueprint-pocketbase-react/a2app.toolkit.json index 3bb782d..7aacf80 100644 --- a/toolkits/blueprint-pocketbase-react/a2app.toolkit.json +++ b/toolkits/blueprint-pocketbase-react/a2app.toolkit.json @@ -8,7 +8,7 @@ "gate": [ { "name": "hooks syntax + rules self-test", - "run": "node --check pb/pb_hooks/_a2app_rules.js && node --check pb/pb_hooks/_a2app_impl.js && node --check pb/pb_hooks/_a2app.pb.js && node pb/pb_hooks/_a2app_rules.js --selftest" + "run": "node --check pb/pb_hooks/_a2app_rules.js && node --check pb/pb_hooks/_a2app_impl.js && node --check pb/pb_hooks/_a2app.pb.js && node pb/pb_hooks/_a2app_rules.js --selftest && node scripts/test-triage.cjs" }, { "name": "operations resolve (every declared op has a runner)", diff --git a/toolkits/blueprint-pocketbase-react/package.json b/toolkits/blueprint-pocketbase-react/package.json index 379d79e..d5352ee 100644 --- a/toolkits/blueprint-pocketbase-react/package.json +++ b/toolkits/blueprint-pocketbase-react/package.json @@ -9,6 +9,6 @@ "scripts": { "build": "echo \"blueprint — nothing to build yet\" && exit 0", "typecheck": "echo \"blueprint — nothing to typecheck yet\" && exit 0", - "test": "echo \"no tests yet\" && exit 0" + "test": "node --check template/pb/pb_hooks/_a2app_impl.js && node test/rules.test.cjs && node template/scripts/test-triage.cjs" } } diff --git a/toolkits/blueprint-pocketbase-react/template/pb/pb_hooks/_a2app_impl.js b/toolkits/blueprint-pocketbase-react/template/pb/pb_hooks/_a2app_impl.js index 9e777ff..8fcca1e 100644 --- a/toolkits/blueprint-pocketbase-react/template/pb/pb_hooks/_a2app_impl.js +++ b/toolkits/blueprint-pocketbase-react/template/pb/pb_hooks/_a2app_impl.js @@ -1064,6 +1064,7 @@ function runOperation(e, name) { $app.runInTransaction((tx) => { const a2app = { app: tx, + getTask: (id) => loadTask(tx.db(), id), trigger: (type, payload, capability) => triggerWith(tx, env, type, payload, capability), error: operationError, }; diff --git a/toolkits/blueprint-pocketbase-react/template/pb/pb_hooks/operations.js b/toolkits/blueprint-pocketbase-react/template/pb/pb_hooks/operations.js index b403f58..e140a2d 100644 --- a/toolkits/blueprint-pocketbase-react/template/pb/pb_hooks/operations.js +++ b/toolkits/blueprint-pocketbase-react/template/pb/pb_hooks/operations.js @@ -14,7 +14,7 @@ * * args the guarded arguments, as plain JS. * ctx who is calling: { credentialId, agentName, principal }. - * a2app { app, trigger, error }: + * a2app { app, trigger, getTask, error }: * app the PocketBase app to read and write through — * `a2app.app.findRecordById(...)`, `a2app.app.save(record)`. * It is a TRANSACTION: everything the runner writes, and every @@ -25,7 +25,8 @@ * a capability is named, queue a task for an agent. Returns * { eventId, taskId }. An event type missing from `events` * below is refused. - * error (status, code, message) — `throw a2app.error(409, "already_done", + * getTask (id) — read the agent task inside this transaction, or null. + * error (status, code, message, extra?) — `throw a2app.error(409, "already_done", * "…")` refuses with your own status and code. * * Return a plain, JSON-able value; it is the `result` of the call. Anything a @@ -91,9 +92,8 @@ module.exports = { // // Identical triggers dedupe to ONE task, even after it has finished, so // asking again with the same payload would hand back the old failure. - // Naming the previous task makes each request a new occurrence. The View - // disables the control while a run is open, so a double click cannot queue - // two. + // Naming the previous task makes each request a new occurrence. The runner + // must refuse while that task is open, including CLI and second-tab calls. // // The task id goes on the record so the View can show the work until it is // done (ui/src/AgentTask.jsx). The validate gate checks that it does. @@ -102,7 +102,14 @@ module.exports = { if (!task) return { ok: false, reason: "no such task" }; const payload = { task: task.id }; const previous = task.getString("agentTask"); - if (previous) payload.previous = previous; + if (previous) { + const current = a2app.getTask(previous); + if (!current) throw a2app.error(409, "agent_task_unavailable", "The previous agent task could not be found.", { taskId: previous }); + if (["completed", "failed", "canceled"].indexOf(current.status) < 0) { + throw a2app.error(409, "already_queued", "This record already has unfinished agent work.", { taskId: previous }); + } + payload.previous = previous; + } const fired = a2app.trigger("task.needs_triage", payload, "triage"); task.set("agentTask", fired.taskId); a2app.app.save(task); diff --git a/toolkits/blueprint-pocketbase-react/template/reference/blueprint.md b/toolkits/blueprint-pocketbase-react/template/reference/blueprint.md index a15d945..637762c 100644 --- a/toolkits/blueprint-pocketbase-react/template/reference/blueprint.md +++ b/toolkits/blueprint-pocketbase-react/template/reference/blueprint.md @@ -224,7 +224,15 @@ runners: { "request-triage": (args, _ctx, a2app) => { const task = a2app.app.findRecordById("tasks", args.task); const payload = { task: task.id }; - if (task.getString("agentTask")) payload.previous = task.getString("agentTask"); + const previous = task.getString("agentTask"); + if (previous) { + const current = a2app.getTask(previous); + if (!current) throw a2app.error(409, "agent_task_unavailable", "Previous task missing.", { taskId: previous }); + if (["completed", "failed", "canceled"].indexOf(current.status) < 0) { + throw a2app.error(409, "already_queued", "Unfinished agent work exists.", { taskId: previous }); + } + payload.previous = previous; + } const { taskId } = a2app.trigger("task.needs_triage", payload, "triage"); task.set("agentTask", taskId); a2app.app.save(task); @@ -240,6 +248,14 @@ something — nothing is queued and no agent is ever handed it. - **The type must be in `events`.** The adapter refuses an undeclared type at fire time, inside the operation, so a typo works in review and fails in front of a user — the operation answers `operation_failed` and rolls back. +- **Refuse while the record has unfinished agent work.** `a2app.getTask(id)` + reads its queue task in the same transaction as `a2app.trigger` and record + writes. `submitted`, `working` and `input-required` answer 409 + `already_queued` with `taskId`, before any event or write. Only terminal tasks + (`completed`, `failed`, `canceled`) permit a new occurrence with `previous`. + A missing task answers 409 `agent_task_unavailable`; failed reads also refuse. + Throw `a2app.error(status, code, message, extra)` for a deliberate refusal. + A disabled View control alone cannot guard CLI calls or a second tab. - **Send ids.** The agent re-reads the record; a copy in the payload is stale by the time it is read, and prose there is an instruction the app does not get to give. diff --git a/toolkits/blueprint-pocketbase-react/template/scripts/test-triage.cjs b/toolkits/blueprint-pocketbase-react/template/scripts/test-triage.cjs new file mode 100644 index 0000000..ac17f8a --- /dev/null +++ b/toolkits/blueprint-pocketbase-react/template/scripts/test-triage.cjs @@ -0,0 +1,33 @@ +const assert = require("node:assert/strict"); +const fs = require("node:fs"); +const path = require("node:path"); +const vm = require("node:vm"); +const loaded = { exports: {} }; +vm.runInNewContext(fs.readFileSync(path.join(__dirname, "../pb/pb_hooks/operations.js"), "utf8"), { module: loaded }); +const { runners } = loaded.exports; + +for (const state of ["submitted", "working", "input-required", "completed", "failed", "canceled", null]) { + const record = { id: "record", agentTask: "old", getString(key) { return this[key]; }, set(key, value) { this[key] = value; } }; + let triggers = 0, saves = 0; + const toolbox = { + app: { findRecordById: () => record, save: () => saves++ }, + getTask: () => state === null ? null : { status: state }, + error: (status, code, message, extra) => ({ a2appOperationError: true, status, code, message, extra }), + trigger: (type, payload) => { triggers++; assert.equal(payload.previous, "old"); return { taskId: "new" }; }, + }; + if (["completed", "failed", "canceled"].includes(state)) { + assert.equal(runners["request-triage"]({ task: "record" }, {}, toolbox).queued, "new"); + assert.equal(record.agentTask, "new"); + assert.equal(triggers, 1); + assert.equal(saves, 1); + } else { + assert.throws(() => runners["request-triage"]({ task: "record" }, {}, toolbox), + (e) => e.status === 409 && e.code === (state === null ? "agent_task_unavailable" : "already_queued") && e.extra.taskId === "old"); + assert.equal(triggers, 0); + assert.equal(saves, 0); + assert.equal(record.agentTask, "old"); + } +} +const source = fs.readFileSync(path.join(__dirname, "../pb/pb_hooks/_a2app_impl.js"), "utf8"); +assert.match(source, /getTask:\s*\(id\)\s*=>\s*loadTask\(tx\.db\(\), id\)/); +console.log("triage test: refusal, terminal retries, no side effects, transaction task read pass"); diff --git a/toolkits/blueprint-pocketbase-react/test/rules.test.cjs b/toolkits/blueprint-pocketbase-react/test/rules.test.cjs new file mode 100644 index 0000000..3e3befd --- /dev/null +++ b/toolkits/blueprint-pocketbase-react/test/rules.test.cjs @@ -0,0 +1,8 @@ +// PocketBase hooks are CommonJS/Goja source inside an ESM workspace package. +const { readFileSync } = require("node:fs"); +const { join } = require("node:path"); +const { runInNewContext } = require("node:vm"); +process.argv.push("--selftest"); +runInNewContext(readFileSync(join(__dirname, "../template/pb/pb_hooks/_a2app_rules.js"), "utf8"), { + module: { exports: {} }, require, process, console, +}); diff --git a/toolkits/blueprint-python-fastapi/a2app.toolkit.json b/toolkits/blueprint-python-fastapi/a2app.toolkit.json index 975d141..c1226f6 100644 --- a/toolkits/blueprint-python-fastapi/a2app.toolkit.json +++ b/toolkits/blueprint-python-fastapi/a2app.toolkit.json @@ -15,7 +15,7 @@ "gate": [ { "name": "python syntax + adapter self-test", - "run": "python -m py_compile a2app_adapter.py main.py schema.py scripts/promote_apply.py && python a2app_adapter.py --selftest" + "run": "python -m py_compile a2app_adapter.py main.py schema.py scripts/promote_apply.py && python a2app_adapter.py --selftest && python -m unittest test_triage" }, { "name": "operations resolve (every declared op has a runner)", diff --git a/toolkits/blueprint-python-fastapi/template/a2app_adapter.py b/toolkits/blueprint-python-fastapi/template/a2app_adapter.py index 9425c40..559887e 100644 --- a/toolkits/blueprint-python-fastapi/template/a2app_adapter.py +++ b/toolkits/blueprint-python-fastapi/template/a2app_adapter.py @@ -484,6 +484,14 @@ def check(self, caller: str, cls: str) -> dict: ) +class OperationError(Exception): + """A runner's explicit HTTP refusal; ordinary exceptions remain 500s.""" + + def __init__(self, status: int, code: str, message: str, **extra): + super().__init__(message) + self.status, self.code, self.extra = status, code, extra + + class UnsupportedFilter(Exception): """Raised when a `filter` expression falls outside the grammar above.""" @@ -564,6 +572,8 @@ def __init__(self, seed: Optional[dict] = None): self.dedup: dict[str, str] = {} self._task_seq = 0 self._event_seq = 0 + # Serialize whole runners, not just individual database calls. + self.operation_lock = threading.Lock() # records --------------------------------------------------------------- def _all_records(self, entity: str) -> list[dict]: @@ -1597,8 +1607,11 @@ def _handle_operation(self, headers, name, args): if not runner: return self._err(501, "not_implemented", f'This app declares "{name}" but implements no operation runner.') try: - result = runner(args, ctx, self.store) + with self.store.operation_lock: + result = runner(args, ctx, self.store) return 200, {"a2app": True, "ok": True, "operation": name, "result": result} + except OperationError as e: + return self._err(e.status, e.code, str(e), **e.extra) except Exception as e: # noqa: BLE001 return self._err(500, "operation_failed", f'Operation "{name}" threw: {e}') diff --git a/toolkits/blueprint-python-fastapi/template/reference/blueprint.md b/toolkits/blueprint-python-fastapi/template/reference/blueprint.md index 345523a..dc480ae 100644 --- a/toolkits/blueprint-python-fastapi/template/reference/blueprint.md +++ b/toolkits/blueprint-python-fastapi/template/reference/blueprint.md @@ -182,6 +182,16 @@ capability is named, puts a task on the app's queue; it returns something — nothing is queued and no agent is handed it. The starter's `request-triage` operation is the worked example. +- **Refuse while the record has unfinished agent work.** Read its current + queue task with `store.get_task(id)`. `submitted`, `working` and + `input-required` answer HTTP 409 `already_queued`, naming `taskId`, before + emitting an event or writing anything. Only terminal tasks (`completed`, + `failed`, `canceled`) permit a re-ask with `previous`. A missing task answers + 409 `agent_task_unavailable`; failed reads also refuse. + The adapter serializes complete runners with a lock. Keep the check, + enqueue and record update in one runner. A disabled View control cannot guard CLI calls + or another tab. Raise/return `OperationError(409, "already_queued", "...", taskId=id)` + for a structured refusal; ordinary errors remain 500 `operation_failed`. - **Declare the type in `EVENTS` first.** A type that is not declared is refused at the moment of firing — inside the operation, in front of a user. - **Firing the same occurrence twice makes one task.** The same type, diff --git a/toolkits/blueprint-python-fastapi/template/schema.py b/toolkits/blueprint-python-fastapi/template/schema.py index 2924a08..dd4c3ae 100644 --- a/toolkits/blueprint-python-fastapi/template/schema.py +++ b/toolkits/blueprint-python-fastapi/template/schema.py @@ -10,6 +10,8 @@ every module has no screen and cannot be reached by walking. """ +from a2app_adapter import OperationError + ENTITIES = { "tasks": { "module": "planning", @@ -118,6 +120,11 @@ def _request_triage(args, ctx, store): return {"ok": False, "reason": "no such task"} payload = {"task": task["id"]} if task.get("agentTask"): + current = store.get_task(task["agentTask"]) + if current is None: + raise OperationError(409, "agent_task_unavailable", "The previous agent task could not be found.", taskId=task["agentTask"]) + if current["status"] not in ("completed", "failed", "canceled"): + raise OperationError(409, "already_queued", "This record already has unfinished agent work.", taskId=task["agentTask"]) payload["previous"] = task["agentTask"] fired = store.trigger("task.needs_triage", payload, capability="triage") task["agentTask"] = fired["taskId"] diff --git a/toolkits/blueprint-python-fastapi/template/test_triage.py b/toolkits/blueprint-python-fastapi/template/test_triage.py new file mode 100644 index 0000000..32760f1 --- /dev/null +++ b/toolkits/blueprint-python-fastapi/template/test_triage.py @@ -0,0 +1,78 @@ +"""Starter regression: run through the adapter, not just the runner.""" +import copy +import time +import unittest +from concurrent.futures import ThreadPoolExecutor + +import schema +from a2app_adapter import Adapter, Store + + +def fixture(): + store = Store(schema.SEED) + adapter = Adapter(app_id="triage-test", app_name="test", entities=schema.ENTITIES, + operations=schema.OPERATIONS, store=store, token="test", + modules=[{"name": "planning"}], operation_runners=schema.OPERATION_RUNNERS, + events=schema.EVENTS) + return adapter, store + + +def ask(adapter): + return adapter.dispatch("POST", "/api/ops/request-triage", {"x-a2app-token": "test"}, {"task": "task_welcome"}) + + +class TriageTest(unittest.TestCase): + def test_states(self): + for state in ("submitted", "working", "input-required", "completed", "failed", "canceled"): + with self.subTest(state=state): + adapter, store = fixture() + status, body = ask(adapter) + self.assertEqual(status, 200) + previous = body["result"]["queued"] + task = store.get_task(previous) + task["status"] = state + store.save_task(task) + before = copy.deepcopy(store.get_record("tasks", "task_welcome")) + event_count = len(store.events) + status, body = ask(adapter) + if state in ("submitted", "working", "input-required"): + self.assertEqual((status, body["code"], body["taskId"]), (409, "already_queued", previous)) + self.assertEqual(len(store.tasks), 1) + self.assertEqual(len(store.events), event_count) + self.assertEqual(store.get_record("tasks", "task_welcome"), before) + else: + self.assertEqual(status, 200) + new_id = body["result"]["queued"] + self.assertNotEqual(new_id, previous) + self.assertEqual(store.get_task(new_id)["request"]["payload"]["previous"], previous) + + def test_concurrent_requests(self): + adapter, store = fixture() + trigger = store.trigger + def delayed(*args, **kwargs): + time.sleep(0.02) # release the GIL between reading the record and enqueueing + return trigger(*args, **kwargs) + store.trigger = delayed + with ThreadPoolExecutor(max_workers=2) as pool: + replies = list(pool.map(lambda _: ask(adapter), range(2))) + self.assertEqual(sorted(r[0] for r in replies), [200, 409]) + self.assertEqual(len(store.tasks), 1) + self.assertEqual(len(store.events), 1) + + def test_unavailable_queue(self): + adapter, store = fixture() + record = store.get_record("tasks", "task_welcome") + record["agentTask"] = "missing" + store.put_record("tasks", record) + status, body = ask(adapter) + self.assertEqual((status, body["code"], body["taskId"]), (409, "agent_task_unavailable", "missing")) + def failed_read(_id): + raise RuntimeError("queue read failed") + store.get_task = failed_read + status, body = ask(adapter) + self.assertEqual((status, body["code"]), (500, "operation_failed")) + self.assertEqual((len(store.tasks), len(store.events)), (0, 0)) + + +if __name__ == "__main__": + unittest.main() diff --git a/toolkits/blueprint-rails-vue/a2app.toolkit.json b/toolkits/blueprint-rails-vue/a2app.toolkit.json index 915e447..c640abc 100644 --- a/toolkits/blueprint-rails-vue/a2app.toolkit.json +++ b/toolkits/blueprint-rails-vue/a2app.toolkit.json @@ -20,7 +20,7 @@ "gate": [ { "name": "ruby syntax + adapter self-test", - "run": "ruby -c lib/a2app_adapter.rb && ruby -c lib/a2app_schema.rb && ruby -c app/controllers/a2app_controller.rb && ruby lib/a2app_adapter.rb --selftest" + "run": "ruby -c lib/a2app_adapter.rb && ruby -c lib/a2app_schema.rb && ruby -c app/controllers/a2app_controller.rb && ruby lib/a2app_adapter.rb --selftest && ruby test/triage_test.rb" }, { "name": "operations resolve (every declared op has a runner)", diff --git a/toolkits/blueprint-rails-vue/template/lib/a2app_adapter.rb b/toolkits/blueprint-rails-vue/template/lib/a2app_adapter.rb index 763b8ec..d4890a1 100644 --- a/toolkits/blueprint-rails-vue/template/lib/a2app_adapter.rb +++ b/toolkits/blueprint-rails-vue/template/lib/a2app_adapter.rb @@ -27,6 +27,15 @@ require "fileutils" module A2appAdapter + class OperationError < StandardError + attr_reader :status, :code, :extra + + def initialize(status, code, message, extra = {}) + super(message) + @status, @code, @extra = status, code, extra + end + end + RULES_VERSION = "0.1.0" PROTOCOL_VERSION = "0.1" ADAPTER_VERSION = "0.1.0" @@ -867,6 +876,7 @@ def initialize(app_id:, app_name:, entities:, operations:, store:, token:, @auth_mode = auth_mode @allowed_origins = (allowed_origins || []).to_h { |o| [o, true] } @runners = operation_runners || {} + @operation_lock = Mutex.new @credential_hint = credential_hint || "Read the app's .agent-token file (mode 0600) in the project directory." @env = env # Deliberate small extension over the python oracle: dispatch owns the @@ -1633,8 +1643,10 @@ def handle_operation(headers, name, args) return err(501, "not_implemented", "This app declares \"#{name}\" but implements no operation runner.") end begin - result = runner.call(args, ctx, @store) + result = @operation_lock.synchronize { runner.call(args, ctx, @store) } [200, { "a2app" => true, "ok" => true, "operation" => name, "result" => result }] + rescue OperationError => e + err(e.status, e.code, e.message, e.extra) rescue StandardError => e err(500, "operation_failed", "Operation \"#{name}\" threw: #{e.message}") end diff --git a/toolkits/blueprint-rails-vue/template/lib/a2app_schema.rb b/toolkits/blueprint-rails-vue/template/lib/a2app_schema.rb index c53e934..24c9ea8 100644 --- a/toolkits/blueprint-rails-vue/template/lib/a2app_schema.rb +++ b/toolkits/blueprint-rails-vue/template/lib/a2app_schema.rb @@ -130,9 +130,8 @@ module A2appSchema # # Identical triggers dedupe to ONE task, even after it has finished, so # asking again with the same payload would hand back the old failure. - # Naming the previous task makes each request a new occurrence. The View - # disables the control while a run is open, so a double click cannot - # queue two. + # Naming the previous task makes each request a new occurrence. The runner + # must refuse while that task is open, including CLI and second-tab calls. # # The task id goes on the record so the View can show the work until it # is done (ui/src/components/AgentTaskPanel.vue). The validate gate checks @@ -141,7 +140,14 @@ module A2appSchema task = store.get_record("tasks", args["task"]) next { "ok" => false, "reason" => "no such task" } if task.nil? payload = { "task" => task["id"] } - payload["previous"] = task["agentTask"] if task["agentTask"].is_a?(String) && !task["agentTask"].empty? + if task["agentTask"].is_a?(String) && !task["agentTask"].empty? + current = store.get_task(task["agentTask"]) + raise A2appAdapter::OperationError.new(409, "agent_task_unavailable", "The previous agent task could not be found.", { "taskId" => task["agentTask"] }) if current.nil? + unless %w[completed failed canceled].include?(current["status"]) + raise A2appAdapter::OperationError.new(409, "already_queued", "This record already has unfinished agent work.", { "taskId" => task["agentTask"] }) + end + payload["previous"] = task["agentTask"] + end fired = store.trigger("task.needs_triage", payload, "triage") task["agentTask"] = fired["taskId"] store.put_record("tasks", task) diff --git a/toolkits/blueprint-rails-vue/template/reference/blueprint.md b/toolkits/blueprint-rails-vue/template/reference/blueprint.md index e83e8a6..04900a4 100644 --- a/toolkits/blueprint-rails-vue/template/reference/blueprint.md +++ b/toolkits/blueprint-rails-vue/template/reference/blueprint.md @@ -250,6 +250,16 @@ capability is named, puts a task on the app's queue; it returns nothing is queued and no agent is handed it. An undeclared type raises. The starter's `request-triage` operation is the worked example. +- **Refuse while the record has unfinished agent work.** Read its current + queue task with `store.get_task(id)`. `submitted`, `working` and + `input-required` answer HTTP 409 `already_queued`, naming `taskId`, before + emitting an event or writing anything. Only terminal tasks (`completed`, + `failed`, `canceled`) permit a re-ask with `previous`. A missing task answers + 409 `agent_task_unavailable`; failed reads also refuse. + The adapter serializes complete runners with a mutex. Keep the check, + enqueue and record update in one runner. A disabled View control cannot guard CLI calls + or another tab. Raise/return `A2appAdapter::OperationError.new(409, "already_queued", "...", { "taskId" => id })` + for a structured refusal; ordinary errors remain 500 `operation_failed`. - **Declare the type in `EVENTS` first.** A type that is not declared is refused at the moment of firing — inside the operation, in front of a user. - **Firing the same occurrence twice makes one task.** The same type, diff --git a/toolkits/blueprint-rails-vue/template/test/triage_test.rb b/toolkits/blueprint-rails-vue/template/test/triage_test.rb new file mode 100644 index 0000000..d67e492 --- /dev/null +++ b/toolkits/blueprint-rails-vue/template/test/triage_test.rb @@ -0,0 +1,60 @@ +require_relative "../lib/a2app_adapter" +require_relative "../lib/a2app_schema" + +def fixture + store = A2appAdapter::Store.new(A2appSchema::SEED) + app = A2appAdapter::Adapter.new(app_id: "test", app_name: "test", entities: A2appSchema::ENTITIES, + operations: A2appSchema::OPERATIONS, store: store, token: "test", modules: [{ "name" => "planning" }], + operation_runners: A2appSchema::OPERATION_RUNNERS, events: A2appSchema::EVENTS) + [app, store] +end + +def ask(app) + app.dispatch("POST", "/api/ops/request-triage", { "x-a2app-token" => "test" }, { "task" => "task_welcome" }, {}) +end + +%w[submitted working input-required completed failed canceled].each do |state| + app, store = fixture + status, body = ask(app) + raise "initial request: #{body}" unless status == 200 + previous = body["result"]["queued"] + task = store.get_task(previous) + task["status"] = state + store.save_task(task) + before = JSON.generate(store.get_record("tasks", "task_welcome")) + events = store.events_since(nil)["events"].length + status, body = ask(app) + if %w[completed failed canceled].include?(state) + raise "retry: #{body}" unless status == 200 + id = body["result"]["queued"] + raise "retry occurrence" unless id != previous && store.get_task(id)["request"]["payload"]["previous"] == previous + else + raise "refusal: #{body}" unless status == 409 && body["code"] == "already_queued" && body["taskId"] == previous + raise "refusal wrote state" unless store.list_tasks(nil).length == 1 && store.events_since(nil)["events"].length == events && JSON.generate(store.get_record("tasks", "task_welcome")) == before + end +end + +app, store = fixture +record = store.get_record("tasks", "task_welcome") +record["agentTask"] = "missing" +store.put_record("tasks", record) +status, body = ask(app) +raise "missing task: #{body}" unless status == 409 && body["code"] == "agent_task_unavailable" && body["taskId"] == "missing" +raise "missing task queued work" unless store.list_tasks(nil).empty? && store.events_since(nil)["events"].empty? +def store.get_task(_id) + raise "queue read failed" +end +status, body = ask(app) +raise "ordinary failure: #{body}" unless status == 500 && body["code"] == "operation_failed" + +app, store = fixture +class << store + alias original_trigger trigger + def trigger(*args) + sleep 0.02 + original_trigger(*args) + end +end +replies = 2.times.map { Thread.new { ask(app) } }.map(&:value) +raise "concurrent requests" unless replies.map(&:first).sort == [200, 409] && store.list_tasks(nil).length == 1 +puts "triage test: refusal, terminal retries, no side effects, concurrent requests pass" diff --git a/toolkits/blueprint-react-node/package.json b/toolkits/blueprint-react-node/package.json index b263675..8d0dc64 100644 --- a/toolkits/blueprint-react-node/package.json +++ b/toolkits/blueprint-react-node/package.json @@ -11,6 +11,6 @@ "scripts": { "build": "echo \"blueprint \u00e2\u20ac\u201d nothing to build yet\" && exit 0", "typecheck": "echo \"blueprint \u00e2\u20ac\u201d nothing to typecheck yet\" && exit 0", - "test": "node --check template/server.mjs && node --check template/a2app-update.js && node test/update-watch.test.mjs && node test/app-to-agent.test.mjs" + "test": "node --check template/server.mjs && node --check template/a2app-update.js && node test/update-watch.test.mjs && node test/app-to-agent.test.mjs && node test/triage.test.mjs" } -} \ No newline at end of file +} diff --git a/toolkits/blueprint-react-node/template/a2app.schema.mjs b/toolkits/blueprint-react-node/template/a2app.schema.mjs index 6a1fdf3..030c4f9 100644 --- a/toolkits/blueprint-react-node/template/a2app.schema.mjs +++ b/toolkits/blueprint-react-node/template/a2app.schema.mjs @@ -96,7 +96,9 @@ export const schema = { // about. events: [{ type: "task.needs_triage" }], - // Runner signature: (args, ctx, { store, trigger }) => jsonable result. `store` is the + // Runner signature: (args, ctx, { store, trigger, getTask, error }) => JSON. + // `getTask(id)` reads queue state; throw `error(status, code, message, extra)` + // for a deliberate HTTP refusal. `store` is the // SQLite-backed record store — list(entity) · get(entity, id) · put(entity, // record) · remove(entity, id). Writes are durable when the call returns; // there is no separate persist() step on this stack. `trigger(type, payload, @@ -132,15 +134,21 @@ export const schema = { // // Identical triggers dedupe to ONE task, even after it has finished, so // asking again with the same payload would hand back the old failure. - // Naming the previous task makes each request a new occurrence. The View - // disables the control while a run is open, so a double click cannot - // queue two. + // Naming the previous task makes each request a new occurrence. The runner + // must refuse while that task is open, including CLI and second-tab calls. // // The task id goes on the record so the View can show the work until it is // done (src/AgentTask.jsx). The validate gate checks that it does. - "request-triage": (args, _ctx, { store, trigger }) => { + "request-triage": (args, _ctx, { store, trigger, getTask, error }) => { const task = store.get("tasks", args?.task); if (!task) return { ok: false, reason: "no such task" }; + if (task.agentTask) { + const current = getTask(task.agentTask); + if (!current) throw error(409, "agent_task_unavailable", "The previous agent task could not be found.", { taskId: task.agentTask }); + if (!["completed", "failed", "canceled"].includes(current.status)) { + throw error(409, "already_queued", "This record already has unfinished agent work.", { taskId: task.agentTask }); + } + } const payload = task.agentTask ? { task: task.id, previous: task.agentTask } : { task: task.id }; const { taskId } = trigger("task.needs_triage", payload, "triage"); task.agentTask = taskId; diff --git a/toolkits/blueprint-react-node/template/reference/blueprint.md b/toolkits/blueprint-react-node/template/reference/blueprint.md index 6dc6593..317b66c 100644 --- a/toolkits/blueprint-react-node/template/reference/blueprint.md +++ b/toolkits/blueprint-react-node/template/reference/blueprint.md @@ -224,10 +224,20 @@ handle that enqueues agent work is not yours to reach directly — you get it as events: [{ type: "task.needs_triage" }], operationRunners: { - "request-triage": (args, _ctx, { store, trigger }) => { + "request-triage": (args, _ctx, { store, trigger, getTask, error }) => { const task = store.get("tasks", args?.task); if (!task) return { ok: false, reason: "no such task" }; - const { taskId } = trigger("task.needs_triage", { task: task.id }, "triage"); + if (task.agentTask) { + const current = getTask(task.agentTask); + if (!current) throw error(409, "agent_task_unavailable", "Previous task missing.", { taskId: task.agentTask }); + if (!["completed", "failed", "canceled"].includes(current.status)) { + throw error(409, "already_queued", "Unfinished agent work exists.", { taskId: task.agentTask }); + } + } + const payload = task.agentTask ? { task: task.id, previous: task.agentTask } : { task: task.id }; + const { taskId } = trigger("task.needs_triage", payload, "triage"); + task.agentTask = taskId; + store.put("tasks", task); return { ok: true, queued: taskId }; }, }, @@ -242,6 +252,14 @@ something — nothing is queued and no agent is ever handed it. inside the operation, so an undeclared type is a feature that works in review and fails in front of a user. `test/app-to-agent.test.mjs` checks the two agree. +- **Refuse while the record has unfinished agent work.** `getTask(id)` reads + its current queue task. `submitted`, `working` and `input-required` answer + 409 `already_queued` with `taskId`, before any event or write. Terminal tasks + (`completed`, `failed`, `canceled`) permit a new occurrence with `previous`. + A missing task answers 409 `agent_task_unavailable`; failed reads also refuse. + `error(status, code, message, extra)` creates a structured refusal to throw. + Keep check, trigger and record update synchronous in this runner so requests + cannot interleave. A disabled View control alone cannot guard CLI calls. - **Send ids.** The agent re-reads the record; a copy in the payload is stale by the time it is read, and prose there is an instruction the app does not get to give. diff --git a/toolkits/blueprint-react-node/template/server.mjs b/toolkits/blueprint-react-node/template/server.mjs index ad6f377..e9285a7 100644 --- a/toolkits/blueprint-react-node/template/server.mjs +++ b/toolkits/blueprint-react-node/template/server.mjs @@ -10,7 +10,7 @@ * `schemaVersion` are derived from the live schema, an agent always sees the * true model. */ -import { createA2App, createStaticView, UnsupportedFilterError } from "@a2app/adapter-core"; +import { createA2App, createStaticView, OperationError, UnsupportedFilterError } from "@a2app/adapter-core"; import { serve } from "@hono/node-server"; import { RESPONSE_ALREADY_SENT } from "@hono/node-server/utils/response"; import { Hono } from "hono"; @@ -225,6 +225,8 @@ const binding = { // always after boot. return runner(args, ctx, { store, + getTask: (id) => a2app.store.getTask(id), + error: (status, code, message, extra) => new OperationError(code, message, status, extra), /** Emit a declared event, and enqueue a task when `capability` is given. */ trigger: (type, payload, capability) => a2app.trigger({ type, payload: payload ?? {}, ...(capability ? { capability } : {}) }), }); diff --git a/toolkits/blueprint-react-node/test/app-to-agent.test.mjs b/toolkits/blueprint-react-node/test/app-to-agent.test.mjs index b808323..88072df 100644 --- a/toolkits/blueprint-react-node/test/app-to-agent.test.mjs +++ b/toolkits/blueprint-react-node/test/app-to-agent.test.mjs @@ -83,6 +83,8 @@ if (typeof runner === "function") { calls.push({ type, payload, capability }); return { eventId: "ev_1", taskId: "tsk_1" }; }, + getTask: () => ({ status: "working" }), + error: (status, code, message, extra) => Object.assign(new Error(message), { status, code, ...extra }), }; const result = runner({ task: "task_welcome" }, {}, toolbox); diff --git a/toolkits/blueprint-react-node/test/triage-fixture.mjs b/toolkits/blueprint-react-node/test/triage-fixture.mjs new file mode 100644 index 0000000..ce99003 --- /dev/null +++ b/toolkits/blueprint-react-node/test/triage-fixture.mjs @@ -0,0 +1,24 @@ +import { createA2App, MemoryBinding, OperationError } from "../../../adapters/adapter-core/dist/index.js"; +import { schema } from "../template/a2app.schema.mjs"; + +// The real starter runner and protocol adapter, with disposable record storage. +export function triageFixture(token = "triage-test") { + let app; + const binding = new MemoryBinding({ appId: "triage-test", entities: schema.entities }); + const toolbox = { + store: { + get: (entity, id) => structuredClone(binding.getRecord(entity, id)), + put: (entity, record) => Object.assign(binding.getRecord(entity, record.id), record), + }, + trigger: (type, payload, capability) => app.trigger({ type, payload, capability }), + getTask: (id) => app.store.getTask(id), + error: (status, code, message, extra) => new OperationError(code, message, status, extra), + }; + binding.runOperation = (name, args, ctx) => schema.operationRunners[name](args, ctx, toolbox); + app = createA2App(binding, { + operations: schema.operations, events: schema.events, modules: [{ name: "planning" }], + credentials: [{ token, credentialId: "test", principal: "owner", scopes: ["*"] }], + }); + const call = (method, path, body = {}) => app.handle({ method, path, body, query: {}, headers: { "x-a2app-token": token } }); + return { app, binding, toolbox, call, ask: () => call("POST", "/api/ops/request-triage", { task: "task_welcome" }) }; +} diff --git a/toolkits/blueprint-react-node/test/triage.test.mjs b/toolkits/blueprint-react-node/test/triage.test.mjs new file mode 100644 index 0000000..b7a2819 --- /dev/null +++ b/toolkits/blueprint-react-node/test/triage.test.mjs @@ -0,0 +1,51 @@ +import assert from "node:assert/strict"; +import { readFileSync } from "node:fs"; +import { triageFixture } from "./triage-fixture.mjs"; + +for (const status of ["submitted", "working", "input-required", "completed", "failed", "canceled"]) { + const f = triageFixture(); + const first = await f.ask(); + assert.equal(first.status, 200); + const id = first.json.result.queued; + const task = f.app.store.getTask(id); + task.status = status; + f.app.store.saveTask(task); + const before = JSON.stringify(f.binding.getRecord("tasks", "task_welcome")); + const events = f.app.store.eventsSince(null).events.length; + const second = await f.ask(); + if (["submitted", "working", "input-required"].includes(status)) { + assert.equal(second.status, 409, status); + assert.equal(second.json.code, "already_queued"); + assert.equal(second.json.taskId, id); + assert.equal(f.app.store.listTasks().length, 1); + assert.equal(f.app.store.eventsSince(null).events.length, events); + assert.equal(JSON.stringify(f.binding.getRecord("tasks", "task_welcome")), before); + } else { + assert.equal(second.status, 200, status); + assert.notEqual(second.json.result.queued, id); + assert.equal(f.app.store.getTask(second.json.result.queued).request.payload.previous, id); + } +} +{ + const f = triageFixture(); + const results = await Promise.all([f.ask(), f.ask()]); + assert.deepEqual(results.map((r) => r.status).sort(), [200, 409]); + assert.equal(f.app.store.listTasks().length, 1); +} +{ + const f = triageFixture(); + f.binding.getRecord("tasks", "task_welcome").agentTask = "missing"; + assert.equal((await f.ask()).json.code, "agent_task_unavailable"); + assert.equal(f.app.store.listTasks().length, 0); + assert.equal(f.app.store.eventsSince(null).events.length, 0); + f.toolbox.getTask = () => { throw new Error("queue read failed"); }; + const failed = await f.ask(); + assert.equal(failed.status, 500); + assert.equal(failed.json.code, "operation_failed"); + assert.equal(f.app.store.listTasks().length, 0); +} +// Pin the system-owned wiring too: unit fixtures must not mask a missing seam. +const server = readFileSync(new URL("../template/server.mjs", import.meta.url), "utf8"); +assert.match(server, /getTask:\s*\(id\)\s*=>\s*a2app\.store\.getTask\(id\)/); +assert.match(server, /new OperationError\(code, message, status, extra\)/); +console.log("✓ triage: HTTP refusal, terminal retries, no side effects, concurrent requests"); diff --git a/toolkits/blueprint-rust-react/a2app.toolkit.json b/toolkits/blueprint-rust-react/a2app.toolkit.json index 0c4c234..bc62bcc 100644 --- a/toolkits/blueprint-rust-react/a2app.toolkit.json +++ b/toolkits/blueprint-rust-react/a2app.toolkit.json @@ -17,7 +17,7 @@ "gate": [ { "name": "cargo build + adapter self-test", - "run": "cargo run --quiet -- --selftest" + "run": "cargo test --quiet && cargo run --quiet -- --selftest" }, { "name": "operations resolve (every declared op has a runner)", diff --git a/toolkits/blueprint-rust-react/template/reference/blueprint.md b/toolkits/blueprint-rust-react/template/reference/blueprint.md index 0ecf976..fc44fd4 100644 --- a/toolkits/blueprint-rust-react/template/reference/blueprint.md +++ b/toolkits/blueprint-rust-react/template/reference/blueprint.md @@ -128,7 +128,7 @@ pub fn operation_runner(name: &str) -> Option { } } -fn run_complete_task(args: &Value, _ctx: &Value, store: &mut Store) -> Result { +fn run_complete_task(args: &Value, _ctx: &Value, store: &mut Store) -> Result { let id = args.get("task").and_then(Value::as_str).unwrap_or(""); let mut task = match store.get_record("tasks", id) { Some(t) => t, @@ -140,9 +140,11 @@ fn run_complete_task(args: &Value, _ctx: &Value, store: &mut Store) -> Result Result` - — (args, ctx, store) to a JSON-able result; an `Err` surfaces as a 500 - `operation_failed` envelope. `store` is the SQLite-backed record store — +- **Runner signature:** `fn(&Value, &Value, &mut Store) -> Result` + — (args, ctx, store) to a JSON-able result or a structured refusal. String + errors convert with `?` / `.into()` to 500 `operation_failed`. Import + `OperationError` from `crate::a2app_adapter`. `store` is the SQLite-backed + record store — `list_records(entity, query)` · `get_record(entity, id)` · `put_record(entity, record)` · `delete_record(entity, id)`. Writes are durable when the call returns; there is no separate persist() step. @@ -228,7 +230,7 @@ pub fn events() -> Value { json!([{ "type": "invoice.needs_review" }]) // every type a runner fires } -fn run_request_review(args: &Value, _ctx: &Value, store: &mut Store) -> Result { +fn run_request_review(args: &Value, _ctx: &Value, store: &mut Store) -> Result { let id = args.get("invoice").and_then(Value::as_str).unwrap_or(""); let fired = store.trigger("invoice.needs_review", &json!({ "invoice": id }), Some("review"))?; Ok(json!({ "ok": true, "queued": fired["taskId"] })) @@ -241,6 +243,16 @@ capability is given, puts a task on the app's queue; it returns queued and no agent is handed it. An undeclared type is an `Err`. The starter's `request-triage` operation is the worked example. +- **Refuse while the record has unfinished agent work.** Read its current + queue task with `store.get_task(id)`. `submitted`, `working` and + `input-required` answer HTTP 409 `already_queued`, naming `taskId`, before + emitting an event or writing anything. Only terminal tasks (`completed`, + `failed`, `canceled`) permit a re-ask with `previous`. A missing task answers + 409 `agent_task_unavailable`; failed reads also refuse. + The server processes requests serially and each runner exclusively borrows + the store. Keep the check, enqueue and record update in one runner. A disabled View control cannot guard CLI calls + or another tab. Raise/return `OperationError { status: 409, code: "already_queued".into(), message: "...".into(), extra: json!({ "taskId": id }) }` + for a structured refusal; ordinary errors remain 500 `operation_failed`. - **Declare the type in `events()` first.** A type that is not declared is refused at the moment of firing — inside the operation, in front of a user. - **Firing the same occurrence twice makes one task.** The same type, diff --git a/toolkits/blueprint-rust-react/template/src/a2app_adapter.rs b/toolkits/blueprint-rust-react/template/src/a2app_adapter.rs index ad88ea7..baede4b 100644 --- a/toolkits/blueprint-rust-react/template/src/a2app_adapter.rs +++ b/toolkits/blueprint-rust-react/template/src/a2app_adapter.rs @@ -1575,9 +1575,23 @@ impl Access { } } -/// One operation runner: (args, ctx, store) -> JSON-able result or an error -/// string (surfaced as a 500 operation_failed envelope). -pub type Runner = fn(&Value, &Value, &mut Store) -> Result; +/// A runner's explicit refusal, or an ordinary failure converted from String. +#[derive(Debug)] +pub struct OperationError { + pub status: u16, + pub code: String, + pub message: String, + pub extra: Value, +} + +impl From for OperationError { + fn from(message: String) -> Self { + Self { status: 500, code: "operation_failed".into(), message, extra: json!({}) } + } +} + +/// One operation runner. String errors still convert to 500 operation_failed. +pub type Runner = fn(&Value, &Value, &mut Store) -> Result; /// Runner resolution by operation name — `schema::operation_runner`. pub type RunnerLookup = fn(&str) -> Option; @@ -2795,7 +2809,17 @@ impl Adapter { }; match runner(args, &ctx, &mut self.store) { Ok(result) => (200, json!({ "a2app": true, "ok": true, "operation": name, "result": result })), - Err(e) => Self::err(500, "operation_failed", &format!("Operation \"{name}\" threw: {e}")), + Err(e) => { + let extra: Vec<(&str, Value)> = e.extra.as_object() + .map(|m| m.iter().map(|(k, v)| (k.as_str(), v.clone())).collect()) + .unwrap_or_default(); + let message = if e.status == 500 && e.code == "operation_failed" { + format!("Operation \"{name}\" threw: {}", e.message) + } else { + e.message.clone() + }; + Self::err_with(e.status, &e.code, &message, &extra) + }, } } diff --git a/toolkits/blueprint-rust-react/template/src/schema.rs b/toolkits/blueprint-rust-react/template/src/schema.rs index eefb1c8..57a468b 100644 --- a/toolkits/blueprint-rust-react/template/src/schema.rs +++ b/toolkits/blueprint-rust-react/template/src/schema.rs @@ -19,7 +19,7 @@ use serde_json::{json, Value}; -use crate::a2app_adapter::{Runner, Store}; +use crate::a2app_adapter::{OperationError, Runner, Store}; pub fn entities() -> Value { json!({ @@ -110,7 +110,8 @@ pub fn seed() -> Value { }) } -// Operation runners: (args, ctx, store) -> Ok(JSON result) or Err(message). +// Operation runners return JSON or OperationError. String errors convert with +// `?` / `.into()` to ordinary 500 operation_failed responses. // The adapter calls the runner for a declared operation; a destructive op is // gated by approval first. `store` is the live SQLite-backed store — read with // `store.get_record` / `store.list_records`, write with `store.put_record` / @@ -128,7 +129,7 @@ pub fn operation_runner(name: &str) -> Option { } } -fn run_clear_done(_args: &Value, _ctx: &Value, store: &mut Store) -> Result { +fn run_clear_done(_args: &Value, _ctx: &Value, store: &mut Store) -> Result { let listed = store.list_records("tasks", &json!({})).map_err(|e| e.to_string())?; let done_ids: Vec = listed .get("items") @@ -150,12 +151,12 @@ fn run_clear_done(_args: &Value, _ctx: &Value, store: &mut Store) -> Result Result { +fn run_count_tasks(_args: &Value, _ctx: &Value, store: &mut Store) -> Result { let listed = store.list_records("tasks", &json!({})).map_err(|e| e.to_string())?; Ok(json!({ "count": listed.get("totalItems").cloned().unwrap_or(json!(0)) })) } -fn run_complete_task(args: &Value, _ctx: &Value, store: &mut Store) -> Result { +fn run_complete_task(args: &Value, _ctx: &Value, store: &mut Store) -> Result { let task_id = args.get("task").and_then(Value::as_str).unwrap_or(""); let mut task = match store.get_record("tasks", task_id) { Some(t) => t, @@ -180,13 +181,12 @@ fn run_complete_task(args: &Value, _ctx: &Value, store: &mut Store) -> Result Result { +fn run_request_triage(args: &Value, _ctx: &Value, store: &mut Store) -> Result { let task_id = args.get("task").and_then(Value::as_str).unwrap_or(""); let mut task = match store.get_record("tasks", task_id) { Some(t) => t, @@ -194,6 +194,16 @@ fn run_request_triage(args: &Value, _ctx: &Value, store: &mut Store) -> Result Result Adapter { + Adapter::new(AdapterConfig { + app_id: "triage-test".into(), app_name: None, + entities: entities(), operations: operations(), store: Store::memory(seed()), + token: "test".into(), modules: json!([{ "name": "planning" }]), + allowed_origins: vec![], runner_lookup: operation_runner, events: events(), + auth_mode: "none".into(), credential_hint: None, env: None, app_version: None, + }).unwrap() + } + + fn ask(app: &mut Adapter) -> (u16, Value) { + app.dispatch("POST", "/api/ops/request-triage", + &HashMap::from([("x-a2app-token".into(), "test".into())]), + Some(&json!({ "task": "task_welcome" })), &HashMap::new()) + } + + #[test] + fn triage_states() { + for state in ["submitted", "working", "input-required", "completed", "failed", "canceled"] { + let mut app = fixture(); + let (status, body) = ask(&mut app); + assert_eq!(status, 200); + let previous = body["result"]["queued"].as_str().unwrap().to_string(); + let mut task = app.store.get_task(&previous).unwrap(); + task["status"] = json!(state); + app.store.save_task(task); + let before = app.store.get_record("tasks", "task_welcome"); + let events = app.store.events_since(None).0.len(); + let (status, body) = ask(&mut app); + if matches!(state, "completed" | "failed" | "canceled") { + assert_eq!(status, 200, "{state}: {body}"); + let id = body["result"]["queued"].as_str().unwrap(); + assert_ne!(id, previous); + assert_eq!(app.store.get_task(id).unwrap()["request"]["payload"]["previous"], previous); + } else { + assert_eq!(status, 409, "{state}: {body}"); + assert_eq!(body["code"], "already_queued"); + assert_eq!(body["taskId"], previous); + assert_eq!(app.store.list_tasks(None).len(), 1); + assert_eq!(app.store.events_since(None).0.len(), events); + assert_eq!(app.store.get_record("tasks", "task_welcome"), before); + } + } + } + + #[test] + fn triage_unavailable() { + let mut app = fixture(); + let mut record = app.store.get_record("tasks", "task_welcome").unwrap(); + record["agentTask"] = json!("missing"); + app.store.put_record("tasks", record); + let (status, body) = ask(&mut app); + assert_eq!(status, 409); + assert_eq!(body["code"], "agent_task_unavailable"); + assert_eq!(body["taskId"], "missing"); + assert_eq!(app.store.list_tasks(None).len(), 0); + assert_eq!(app.store.events_since(None).0.len(), 0); + } + + #[test] + fn ordinary_runner_failure() { + fn fail(_: &Value, _: &Value, _: &mut Store) -> Result { + Err("ordinary failure".to_string().into()) + } + let mut app = Adapter::new(AdapterConfig { + runner_lookup: |_| Some(fail), + app_id: "test".into(), app_name: None, entities: entities(), operations: operations(), + store: Store::memory(seed()), token: "test".into(), modules: json!([{ "name": "planning" }]), + allowed_origins: vec![], events: events(), auth_mode: "none".into(), + credential_hint: None, env: None, app_version: None, + }).unwrap(); + let (status, body) = ask(&mut app); + assert_eq!(status, 500); + assert_eq!(body["code"], "operation_failed"); + assert_eq!(body["message"], "Operation \"request-triage\" threw: ordinary failure"); + } +}