Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions .github/workflows/ci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -53,3 +53,15 @@ jobs:
bun-version: 1.3.14
- run: bun install --frozen-lockfile
- run: bun run build

test:
name: Test
runs-on: ubuntu-latest
needs: [lint, typecheck]
steps:
- uses: actions/checkout@v7
- uses: oven-sh/setup-bun@v2
with:
bun-version: 1.3.14
- run: bun install --frozen-lockfile
- run: bun run test
73 changes: 73 additions & 0 deletions apps/server/src/agent-runtime.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,73 @@
import { afterEach, beforeEach, describe, expect, it } from "bun:test";
import type { DbRepository } from "@nyxel/db";
import { getDb } from "@nyxel/db";
import { createTestUser, installTestDb } from "@nyxel/db/test-utils";
import { cancelAgentRun } from "./agent-runtime";

async function seedWorkspace(db: DbRepository, path: string) {
const user = createTestUser(path);
const workspace = await db.createWorkspace({ userId: user.id, name: "Test workspace" });
return { user, workspace };
}

async function seedAgent(db: DbRepository, workspaceId: string) {
return db.createAgent({
workspaceId,
name: "Worker",
modelId: "anthropic/claude-fake",
autonomyLevel: "autonomous",
});
}

let ctx: Awaited<ReturnType<typeof installTestDb>>;

beforeEach(async () => {
ctx = await installTestDb();
});

afterEach(async () => {
await ctx.cleanup();
});

describe("cancelAgentRun", () => {
it("marks the run cancelled in the DB even with no in-memory AbortController for it", async () => {
const db = getDb();
const { workspace } = await seedWorkspace(db, ctx.path);
const agent = await seedAgent(db, workspace.id);
const task = await db.createTask({
workspaceId: workspace.id,
assignedAgentId: agent.id,
title: "Orphaned run's task",
instruction: "Do the thing.",
status: "running",
});
// Simulates a run whose owning process died/restarted: this test never
// calls executeManagedTask, so there is no activeRunControllers entry
// for this run in this process — cancelAgentRun must still succeed.
const run = await db.createAgentRun({
workspaceId: workspace.id,
taskId: task.id,
agentId: agent.id,
trigger: "task",
status: "running",
startedAt: new Date(),
workerId: "some-other-process",
});

const cancelled = await cancelAgentRun(run.id);

expect(cancelled.status).toBe("cancelled");
expect(cancelled.cancelRequestedAt).not.toBeNull();
expect(cancelled.completedAt).not.toBeNull();

const persisted = await db.getAgentRun(run.id);
expect(persisted?.status).toBe("cancelled");

const cancelledTask = await db.getTask(task.id);
expect(cancelledTask?.status).toBe("cancelled");
});

it("throws for an unknown run id", async () => {
await expect(cancelAgentRun("does-not-exist")).rejects.toThrow("Unknown agent run");
});
});
72 changes: 57 additions & 15 deletions apps/server/src/agent-runtime.ts
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import { randomUUID } from "node:crypto";
import type { AgentRecord, AgentRunRecord, TaskRecord } from "@nyxel/db";
import { getDb } from "@nyxel/db";
import { getModelCapabilities, streamChat } from "@nyxel/model-providers";
Expand All @@ -24,9 +25,24 @@ export interface ExecutionPlan {
* aborted in-flight (cancelAgentRun still marks it cancelled in the DB). */
const activeRunControllers = new Map<string, AbortController>();

/** Identifies this process as the owner of whatever runs it currently holds
* a lease on — written to agentRun.workerId, purely informational (nothing
* currently reads it back to route work), but gives a stale/orphaned run a
* breadcrumb of which process it was left running in. */
const WORKER_ID = randomUUID();

/** How long a run's lease is valid without a renewed heartbeat before the
* stale-run sweep (see scheduler.ts's checkStaleAgentRuns) considers the
* owning process dead and recovers it. Must be comfortably longer than
* HEARTBEAT_INTERVAL_MS so a single missed tick doesn't cause a false
* positive. */
const LEASE_DURATION_MS = 90_000;
const HEARTBEAT_INTERVAL_MS = 30_000;

/** Aborts the in-flight model call for a run (best-effort — a no-op on the
* abort itself if the run isn't live in this process) and always marks the
* run/task cancelled in the DB so the UI reflects it immediately. */
* run/task cancelled in the DB so the UI reflects it immediately, even if
* the run's owning process isn't this one (see AgentRunRecord.cancelRequestedAt). */
export async function cancelAgentRun(runId: string): Promise<AgentRunRecord> {
activeRunControllers.get(runId)?.abort();

Expand All @@ -37,6 +53,7 @@ export async function cancelAgentRun(runId: string): Promise<AgentRunRecord> {
const updated = await db.updateAgentRun(runId, {
status: "cancelled",
completedAt: new Date(),
cancelRequestedAt: new Date(),
});
if (run.taskId) {
const task = await db.getTask(run.taskId);
Expand Down Expand Up @@ -336,6 +353,7 @@ export async function executeManagedTask(input: {
const task = await db.getTask(input.taskId);
if (!task) throw new Error(`Unknown task: ${input.taskId}`);

const startedAt = new Date();
const run = await db.createAgentRun({
workspaceId: task.workspaceId,
taskId: task.id,
Expand All @@ -345,7 +363,10 @@ export async function executeManagedTask(input: {
trigger: input.trigger,
modelId: effectiveModelId(input.agent, task),
status: "running",
startedAt: new Date(),
startedAt,
workerId: WORKER_ID,
heartbeatAt: startedAt,
leaseUntil: new Date(startedAt.getTime() + LEASE_DURATION_MS),
});
await emitNyxelEvent({
workspaceId: task.workspaceId,
Expand All @@ -358,6 +379,30 @@ export async function executeManagedTask(input: {
const controller = new AbortController();
activeRunControllers.set(run.id, controller);

// Renews the run's lease periodically so the stale-run sweep
// (scheduler.ts's checkStaleAgentRuns) doesn't mistake a genuinely
// in-progress run for one whose owning process died. Also the DB-visible
// fallback for cross-process cancellation: if this run were ever picked
// up by another process's cancelAgentRun call (cancelRequestedAt set
// without this process's activeRunControllers entry being reachable),
// the next heartbeat tick notices and aborts locally too.
const heartbeatTimer = setInterval(() => {
void (async () => {
const current = await db.getAgentRun(run.id).catch(() => null);
if (current?.cancelRequestedAt && !controller.signal.aborted) {
controller.abort();
return;
}
await db
.updateAgentRun(run.id, {
heartbeatAt: new Date(),
leaseUntil: new Date(Date.now() + LEASE_DURATION_MS),
})
.catch(() => {});
})();
}, HEARTBEAT_INTERVAL_MS);
if (typeof heartbeatTimer.unref === "function") heartbeatTimer.unref();

// Autonomy Budgets v1 (see ./autonomy-budget.ts) — the tool-call-boundary
// runtime check inside buildToolsForAgent only fires between tool calls,
// so a run that never calls a tool (or sits inside one very long model
Expand All @@ -381,9 +426,7 @@ export async function executeManagedTask(input: {
completedAt: new Date(),
})
.catch(() => {});
await db
.updateTask(task.id, { status: "blocked", errorMessage: reason })
.catch(() => {});
await db.updateTask(task.id, { status: "blocked", errorMessage: reason }).catch(() => {});
await db
.createTaskEvent({
taskId: task.id,
Expand Down Expand Up @@ -428,6 +471,7 @@ export async function executeManagedTask(input: {
throw err;
} finally {
if (budgetTimeout) clearTimeout(budgetTimeout);
clearInterval(heartbeatTimer);
activeRunControllers.delete(run.id);
}
}
Expand Down Expand Up @@ -607,18 +651,18 @@ async function runManagedTask(
// Detected from the tracker's own state rather than string-matching the
// model's final text (unlike the two checks above) — more robust, since
// it doesn't depend on the model actually repeating the tool result back.
const pausedOnBudget =
!pausedOnQuestion && !pausedOnApproval && budgetBlockedReason !== null;
const pausedOnBudget = !pausedOnQuestion && !pausedOnApproval && budgetBlockedReason !== null;
const isPaused = pausedOnQuestion || pausedOnApproval || pausedOnBudget;
// AgentRunStatus has no "blocked" state (a question-pause is still an
// agent run that's technically waiting on the human) — only TaskStatus
// distinguishes "blocked" (question or budget) from "waiting_approval"
// (tool approval).
const taskStatus = pausedOnQuestion || pausedOnBudget
? ("blocked" as const)
: pausedOnApproval
? ("waiting_approval" as const)
: ("completed" as const);
const taskStatus =
pausedOnQuestion || pausedOnBudget
? ("blocked" as const)
: pausedOnApproval
? ("waiting_approval" as const)
: ("completed" as const);

run = await db.updateAgentRun(run.id, {
status: isPaused ? "waiting_approval" : "completed",
Expand All @@ -631,9 +675,7 @@ async function runManagedTask(
status: taskStatus,
resultSummary: output,
completedAt: isPaused ? null : new Date(),
...(pausedOnBudget
? { errorMessage: `Autonomy budget exceeded: ${budgetBlockedReason}` }
: {}),
...(pausedOnBudget ? { errorMessage: `Autonomy budget exceeded: ${budgetBlockedReason}` } : {}),
});
await db.createTaskEvent({
taskId: task.id,
Expand Down
90 changes: 90 additions & 0 deletions apps/server/src/audit.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,90 @@
import { describe, expect, test } from "bun:test";
import { sanitizeForAudit } from "./audit";

describe("sanitizeForAudit", () => {
test("passes through undefined/null unchanged", () => {
expect(sanitizeForAudit(undefined)).toBeUndefined();
expect(sanitizeForAudit(null)).toBeNull();
});

test("redacts top-level secret-shaped keys, case-insensitively", () => {
const sanitized = sanitizeForAudit({
apiKey: "sk-live-abc",
ApiKey: "sk-live-def",
API_KEY: "sk-live-ghi",
token: "tok-live",
accessToken: "at-live",
refreshToken: "rt-live",
Authorization: "Bearer xyz",
password: "hunter2",
secret: "shh",
cookie: "session=abc",
label: "keep me",
}) as Record<string, unknown>;

expect(sanitized.apiKey).toBe("[REDACTED]");
expect(sanitized.ApiKey).toBe("[REDACTED]");
expect(sanitized.API_KEY).toBe("[REDACTED]");
expect(sanitized.token).toBe("[REDACTED]");
expect(sanitized.accessToken).toBe("[REDACTED]");
expect(sanitized.refreshToken).toBe("[REDACTED]");
expect(sanitized.Authorization).toBe("[REDACTED]");
expect(sanitized.password).toBe("[REDACTED]");
expect(sanitized.secret).toBe("[REDACTED]");
expect(sanitized.cookie).toBe("[REDACTED]");
expect(sanitized.label).toBe("keep me");
});

test("redacts secret-shaped keys nested arbitrarily deep", () => {
const sanitized = sanitizeForAudit({
request: {
headers: { authorization: "Bearer live-token" },
body: {
nested: [{ config: { apiKey: "sk-live-nested" } }, { safe: "ok" }],
},
},
}) as {
request: {
headers: { authorization: string };
body: { nested: [{ config: { apiKey: string } }, { safe: string }] };
};
};

expect(sanitized.request.headers.authorization).toBe("[REDACTED]");
expect(sanitized.request.body.nested[0].config.apiKey).toBe("[REDACTED]");
expect(sanitized.request.body.nested[1].safe).toBe("ok");
});

test("does not redact non-secret keys or mutate primitives/arrays", () => {
const sanitized = sanitizeForAudit({
count: 3,
items: ["a", "b", "c"],
nested: { ok: true },
});
expect(sanitized).toEqual({ count: 3, items: ["a", "b", "c"], nested: { ok: true } });
});

test("caps serialized size and marks the result truncated", () => {
const huge = { blob: "x".repeat(50_000) };
const sanitized = sanitizeForAudit(huge) as {
truncated: boolean;
originalChars: number;
preview: string;
};
expect(sanitized.truncated).toBe(true);
expect(sanitized.originalChars).toBeGreaterThan(20_000);
expect(sanitized.preview.length).toBe(20_000);
});

test("leaves small values under the size cap untouched", () => {
const small = { blob: "x".repeat(100) };
const sanitized = sanitizeForAudit(small);
expect(sanitized).toEqual(small);
});

test("redacts secrets even inside a value that also gets truncated", () => {
const huge = { apiKey: "sk-live-secret", blob: "x".repeat(50_000) };
const sanitized = sanitizeForAudit(huge) as { preview: string };
expect(sanitized.preview).not.toContain("sk-live-secret");
});
});
Loading
Loading