fix: fail in progress tasks on crash

CHANGELOG.md

@@ -18,6 +18,7 @@
 - Correctly report A2A messages as failed when AI Core is configured but no AI Core credentials can be found
 - Correctly report input tokens for Anthropic models on AI Core. OpenTelemetry specifies that input_tokens includes cached tokens, but Anthropic did not include the cached tokens in input_tokens
 - Thinking steps are now correctly separated also inside a turn
+- Tasks are now moved into failed state on server crash to avoid quota issue on unstable servers
 - `@agent.llm` is now correctly considered for feature toggled services
 - SAP HANA Cloud based masking is now retried in case of HANA Cloud NLP connectivity issues
 

srv/handlers/graph-executor.js

@@ -19,6 +19,7 @@ import {
   resumeHitl,
   resumeTimeoutHitl,
 } from "./graph-executor/hitl.js"
+import { registerShutdownHook } from "./graph-executor/crash-handler.js"
 
 const LOG = cds.log("agents")
 
@@ -145,6 +146,9 @@ class GraphExecutor {
     this._recursionLimit = options.recursionLimit ?? null
     /** @type {Map<string, AbortController>} per-task abort controllers */
     this._abortControllers = new Map()
+    /** @type {Map<string, string | undefined>} task tenant by task ID */
+    this._taskTenants = new Map()
+    registerShutdownHook(this)
   }
 
   /**
@@ -400,6 +404,7 @@ class GraphExecutor {
     // Cooperative cancellation: per-task AbortController
     const controller = new AbortController()
     this._abortControllers.set(taskId, controller)
+    this._taskTenants.set(taskId, cds.context?.tenant)
 
     // A2A context for tracing
     if (!cds.context) {
@@ -1056,6 +1061,7 @@ class GraphExecutor {
         setSpanAttrs(rootSpan, linkTraceToPrompt())
 
         this._abortControllers.delete(taskId)
+        this._taskTenants.delete(taskId)
         metrics.concurrentExecutions.add(-1, mAttrs)
 
         // Update task record with usage data (non-blocking, best effort)

srv/handlers/graph-executor/crash-handler.js

@@ -0,0 +1,47 @@
+import cds from "@sap/cds"
+
+const LOG = cds.log("agents")
+
+// REVISIT: Check if in the future tasks can be picked up again after restart
+async function markActiveTasksFailed() {
+  const tasksByTenant = new Map()
+
+  for (const executor of registerShutdownHook.executors) {
+    for (const taskId of executor._abortControllers.keys()) {
+      const tenant = executor._taskTenants.get(taskId)
+      const taskIds = tasksByTenant.get(tenant) || []
+      taskIds.push(taskId)
+      tasksByTenant.set(tenant, taskIds)
+    }
+  }
+
+  await Promise.all(
+    [...tasksByTenant].map(async ([tenant, taskIds]) => {
+      const update = () =>
+        UPDATE("cap.agent.Tasks")
+          .where({
+            taskId: { in: [...new Set(taskIds)] },
+            state: { in: ["submitted", "working", "input-required"] },
+          })
+          .set({ state: "failed" })
+
+      if (tenant) return cds.spawn({ tenant, user: cds.User.privileged }, update)
+      return update()
+    }),
+  )
+}
+
+export function registerShutdownHook(executor) {
+  registerShutdownHook.executors.add(executor)
+  if (registerShutdownHook.executors.size > 1) return
+
+  cds.on("shutdown", async () => {
+    try {
+      await markActiveTasksFailed()
+    } catch (err) {
+      LOG.error("Failed to mark active tasks as failed during shutdown", { error: err.message })
+    }
+  })
+}
+
+registerShutdownHook.executors = new Set()

tests/integration/crash-recovery.test.js

@@ -0,0 +1,142 @@
+import { DatabaseSync } from "node:sqlite"
+import { execFile } from "node:child_process"
+import { rmSync } from "node:fs"
+import { randomUUID } from "node:crypto"
+import path from "node:path"
+import { promisify } from "node:util"
+
+import { startServer, stopServer, registerCleanupHandlers } from "../utils/servers.js"
+
+const BOOKSHOP_DIR = path.resolve(import.meta.dirname, "../projects/bookshop")
+const DB_PATH = path.join(BOOKSHOP_DIR, "db.sqlite")
+const PORT = 4700 + Math.floor(Math.random() * 500)
+const ACTIVE_STATES = ["submitted", "working", "input-required"]
+const execFileAsync = promisify(execFile)
+
+let server
+
+function runningTasks() {
+  const db = new DatabaseSync(DB_PATH)
+  try {
+    const placeholders = ACTIVE_STATES.map(() => "?").join(",")
+    return db
+      .prepare(
+        "SELECT taskId, state FROM cap_agent_Tasks WHERE state IN (" +
+          placeholders +
+          ") ORDER BY taskId",
+      )
+      .all(...ACTIVE_STATES)
+  } finally {
+    db.close()
+  }
+}
+
+function taskState(taskId) {
+  const db = new DatabaseSync(DB_PATH)
+  try {
+    return db.prepare("SELECT state FROM cap_agent_Tasks WHERE taskId = ?").get(taskId)?.state
+  } finally {
+    db.close()
+  }
+}
+
+async function waitFor(predicate, timeout = 15_000) {
+  const deadline = Date.now() + timeout
+  let lastValue
+  while (Date.now() < deadline) {
+    // eslint-disable-next-line no-await-in-loop
+    lastValue = await predicate()
+    if (lastValue) return lastValue
+    // eslint-disable-next-line no-await-in-loop
+    await new Promise((resolve) => setTimeout(resolve, 100))
+  }
+  throw new Error("Timed out waiting for condition. Last value: " + JSON.stringify(lastValue))
+}
+
+async function startSlowTask() {
+  const response = await fetch("http://127.0.0.1:" + PORT + "/a2a/slow-agent/", {
+    method: "POST",
+    headers: { "content-type": "application/json", accept: "text/event-stream" },
+    body: JSON.stringify({
+      jsonrpc: "2.0",
+      id: 1,
+      method: "message/stream",
+      params: {
+        message: {
+          kind: "message",
+          messageId: randomUUID(),
+          role: "user",
+          parts: [{ kind: "text", text: "hold until server crash" }],
+        },
+      },
+    }),
+  })
+
+  const reader = response.body.getReader()
+  let buffer = ""
+  while (true) {
+    // eslint-disable-next-line no-await-in-loop
+    const { value, done } = await reader.read()
+    if (done) {
+      throw new Error(
+        "Slow task stream ended before task ID arrived. HTTP " +
+          response.status +
+          ". Body: " +
+          buffer,
+      )
+    }
+    buffer += Buffer.from(value).toString("utf8")
+    const match = buffer.match(/"id"\s*:\s*"([0-9a-f-]{36})"/)
+    if (match) return { taskId: match[1], reader }
+  }
+}
+
+describe("task recovery after server crash", () => {
+  beforeAll(async () => {
+    for (const suffix of ["", "-shm", "-wal"]) {
+      rmSync(DB_PATH + suffix, { force: true })
+    }
+
+    await execFileAsync("npx", ["cds", "deploy", "--to", "sqlite:db.sqlite"], {
+      cwd: BOOKSHOP_DIR,
+      env: { ...process.env, CDS_ENV: "test,crash-test", NODE_ENV: "test" },
+    })
+
+    registerCleanupHandlers(() => {
+      if (server?.exitCode == null) {
+        try {
+          server.kill()
+        } catch {
+          /* process already exited */
+        }
+      }
+    })
+
+    server = await startServer(BOOKSHOP_DIR, PORT, "bookshop crash recovery", {
+      env: { CDS_ENV: "test,crash-test", PORT: String(PORT) },
+    })
+  }, 30_000)
+
+  afterAll(async () => {
+    await stopServer(server, BOOKSHOP_DIR)
+    server = null
+  })
+
+  it("marks running tasks failed across crash and restart", async () => {
+    const { taskId, reader } = await startSlowTask()
+    await waitFor(() => runningTasks().some((task) => task.taskId === taskId))
+
+    await fetch("http://127.0.0.1:" + PORT + "/test/crash").catch(() => undefined)
+    await waitFor(() => server.exitCode != null || runningTasks().length === 0)
+    await waitFor(() => taskState(taskId) === "failed")
+    await waitFor(() => runningTasks().length === 0)
+
+    await reader.cancel().catch(() => undefined)
+    await stopServer(server, BOOKSHOP_DIR, { cleanDb: false })
+    server = await startServer(BOOKSHOP_DIR, PORT, "bookshop crash recovery restart", {
+      env: { CDS_ENV: "test,crash-test", PORT: String(PORT) },
+    })
+
+    expect(runningTasks()).toEqual([])
+  }, 45_000)
+})

tests/projects/bookshop/package.json

@@ -186,6 +186,14 @@
           }
         }
       },
+      "[crash-test]": {
+        "db": {
+          "kind": "sqlite",
+          "credentials": {
+            "url": "db.sqlite"
+          }
+        }
+      },
       "[production]": {
         "auth": "xsuaa"
       },

tests/projects/bookshop/server.js

@@ -0,0 +1,10 @@
+import cds from "@sap/cds"
+
+cds.on("bootstrap", (app) => {
+  app.get("/test/crash", (_req, res) => {
+    res.status(204).end()
+    setImmediate(() => {
+      throw new Error("intentional crash for task recovery test")
+    })
+  })
+})

tests/utils/servers.js

@@ -44,13 +44,15 @@ export function isPortOpen(port) {
  * @param {string} cwd   - Directory of the CAP app to start
  * @param {number} port  - Expected listening port
  * @param {string} label - Human-readable label for error messages
+ * @param {object} [options] - Child process options
+ * @param {object} [options.env] - Environment overrides
  * @returns {Promise<import("node:child_process").ChildProcess>}
  */
-export function startServer(cwd, port, label) {
+export function startServer(cwd, port, label, { env = {} } = {}) {
   return new Promise((resolve, reject) => {
     const proc = spawn("npx", ["cds-serve"], {
       cwd,
-      env: { ...process.env, FORCE_COLOR: "false", NODE_ENV: "development" },
+      env: { ...process.env, FORCE_COLOR: "false", NODE_ENV: "development", ...env },
       stdio: ["ignore", "pipe", "pipe"],
     })
 
@@ -98,8 +100,10 @@ export function startServer(cwd, port, label) {
  *
  * @param {import("node:child_process").ChildProcess|null} proc
  * @param {string} cwd - Directory to clean up SQLite files from
+ * @param {object} [options]
+ * @param {boolean} [options.cleanDb=true] - Remove generated SQLite files
  */
-export async function stopServer(proc, cwd) {
+export async function stopServer(proc, cwd, { cleanDb = true } = {}) {
   if (!proc) return
   if (proc.exitCode == null) {
     proc.kill()
@@ -108,6 +112,7 @@ export async function stopServer(proc, cwd) {
       new Promise((resolve) => setTimeout(resolve, 5_000)),
     ])
   }
+  if (!cleanDb) return
   try {
     for (const f of readdirSync(cwd).filter((f) => /^db.*\.sqlite(-shm|-wal)?$/.test(f))) {
       try {