Compare commits

..

4 commits

Author SHA1 Message Date
Nick Tabeling
a1a0c80acc fix(DIS-106): fix timeout test — wrong YAML key + overly strict schema
Some checks failed
CI / build-and-test (ubuntu-latest) (pull_request) Has been cancelled
CI / build-and-test (windows-latest) (pull_request) Has been cancelled
CI / lint (pull_request) Has been cancelled
Two root causes for the failing timeout test:

1. Test config used `workspace_root` instead of `workspaces_root`, causing
   Zod validation to fail silently, so resolveTimeoutMs() fell back to the
   default 120s — the mock's 200ms close event always beat it.

2. Schema enforced `.int()` on claude_timeout_seconds, rejecting the test's
   0.05s fractional value. Removed the integer constraint since sub-second
   timeouts are valid (and useful for testing).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-10 10:25:21 +02:00
Nick Tabeling
661cf12549 feat(DIS-106): RunResult discriminated union replaces Promise<string>
Add RunResult type in src/agent/types.ts (success|timeout|cli-error|parse-error).
Runner resolves instead of rejects for all error cases.
Router handles all four status variants explicitly.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-10 10:25:21 +02:00
dev
fac6bbd371 Merge pull request 'feat(DIS-103): ChannelQueue serializes messages per channel (FIFO)' (#48) from phase-1/channel-queue into main
Some checks are pending
CI / build-and-test (ubuntu-latest) (push) Waiting to run
CI / build-and-test (windows-latest) (push) Waiting to run
CI / lint (push) Waiting to run
Reviewed-on: #48
2026-04-10 08:23:11 +00:00
Nick Tabeling
8fed430afc feat(DIS-103): ChannelQueue serializes messages per channel (FIFO)
Some checks failed
CI / build-and-test (ubuntu-latest) (pull_request) Has been cancelled
CI / build-and-test (windows-latest) (pull_request) Has been cancelled
CI / lint (pull_request) Has been cancelled
Add ChannelQueue in src/runtime/channel-queue.ts.
Router enqueues each message so same-channel requests run serially.
Different channels remain parallel.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-10 10:17:57 +02:00
4 changed files with 312 additions and 70 deletions

View file

@ -5,8 +5,11 @@ import { DisclawConfig } from "./config/loader";
import { runAgent } from "./agent/runner"; import { runAgent } from "./agent/runner";
import { splitForDiscord } from "./discord/split"; import { splitForDiscord } from "./discord/split";
import { sanitizeForDiscord } from "./runtime/sanitize"; import { sanitizeForDiscord } from "./runtime/sanitize";
import { ChannelQueue } from "./runtime/channel-queue";
import type { RunResult } from "./agent/types"; import type { RunResult } from "./agent/types";
const channelQueue = new ChannelQueue();
/** /**
* Routes a Discord message to the correct agent based on channel_id. * Routes a Discord message to the correct agent based on channel_id.
* Returns true if the message was handled, false if the channel is not * Returns true if the message was handled, false if the channel is not
@ -20,36 +23,30 @@ export async function routeMessage(
): Promise<boolean> { ): Promise<boolean> {
const channelId = message.channelId; const channelId = message.channelId;
// Ignore messages in the management channel (commands only)
if (channelId === managementChannelId) { if (channelId === managementChannelId) {
return false; return false;
} }
// Look up workspace for this channel
const workspace = db.getWorkspaceByChannelId(channelId); const workspace = db.getWorkspaceByChannelId(channelId);
if (!workspace) { if (!workspace) {
return false; return false;
} }
const channel = message.channel as TextChannel; const channel = message.channel as TextChannel;
// Show typing indicator while the agent works
const typingInterval = setInterval(() => {
channel.sendTyping().catch(() => {});
}, 5000);
// Send immediately as well
await channel.sendTyping().catch(() => {});
const sanitizeOpts = { const sanitizeOpts = {
repoRoot: workspace.workspace_path, repoRoot: workspace.workspace_path,
home: os.homedir(), home: os.homedir(),
}; };
await channelQueue.enqueue(channelId, async () => {
const typingInterval = setInterval(() => {
channel.sendTyping().catch(() => {});
}, 5000);
await channel.sendTyping().catch(() => {});
try { try {
// Load recent conversation history for context
const history = db.getConversationHistory(workspace.id, 30); const history = db.getConversationHistory(workspace.id, 30);
// Run the agent
const result: RunResult = await runAgent({ const result: RunResult = await runAgent({
workspacePath: workspace.workspace_path, workspacePath: workspace.workspace_path,
channelName: channel.name, channelName: channel.name,
@ -58,7 +55,6 @@ export async function routeMessage(
}); });
if (result.status === "success") { if (result.status === "success") {
// Store user message
db.addConversation( db.addConversation(
workspace.id, workspace.id,
message.id, message.id,
@ -67,13 +63,7 @@ export async function routeMessage(
message.author.username message.author.username
); );
// Sanitize the full response once, then use the sanitized version
// for both Discord output and conversation history storage.
// This prevents a feedback loop where raw paths in stored history
// get fed back to the agent as context, causing it to reproduce them.
const sanitizedResponse = sanitizeForDiscord(result.text, sanitizeOpts); const sanitizedResponse = sanitizeForDiscord(result.text, sanitizeOpts);
// Split and send response
const chunks = splitForDiscord(sanitizedResponse); const chunks = splitForDiscord(sanitizedResponse);
let firstMsgId: string | null = null; let firstMsgId: string | null = null;
@ -84,7 +74,6 @@ export async function routeMessage(
} }
} }
// Store sanitized assistant response (use first message id for dedup)
if (firstMsgId) { if (firstMsgId) {
db.addConversation( db.addConversation(
workspace.id, workspace.id,
@ -115,6 +104,7 @@ export async function routeMessage(
} finally { } finally {
clearInterval(typingInterval); clearInterval(typingInterval);
} }
});
return true; return true;
} }

View file

@ -0,0 +1,37 @@
type Task = () => Promise<void>;
/**
* Per-channel FIFO task serializer.
*
* Messages arriving in the same Discord channel are enqueued and executed one
* at a time. A failing task logs the error and is swallowed so the next task
* is not blocked. Different channels run their queues in parallel.
*/
export class ChannelQueue {
private queues = new Map<string, Promise<void>>();
enqueue(channelId: string, task: Task): Promise<void> {
// Use the existing tail promise as the predecessor, or a pre-resolved one
// for a fresh channel.
const prev = this.queues.get(channelId) ?? Promise.resolve();
// Chain the new task behind the current tail. Errors are caught and
// swallowed so subsequent tasks always get a chance to run.
const next = prev.then(() => task()).catch((err) => {
console.error(`[ChannelQueue] Task error in channel ${channelId}:`, err);
});
// Register the new tail. Use a finalizer that removes the entry only when
// THIS chain link is still the stored tail — if another task was enqueued
// in the meantime the entry must stay.
this.queues.set(channelId, next);
next.finally(() => {
if (this.queues.get(channelId) === next) {
this.queues.delete(channelId);
}
});
return next;
}
}

View file

@ -0,0 +1,115 @@
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
/**
* Integration test: verifies that routeMessage serializes same-channel messages
* through ChannelQueue so they are processed in FIFO order.
*
* runAgent is mocked to record call order without spawning a real process.
* discord.js and the database are also mocked so no real I/O occurs.
*/
// ---- Mock runAgent BEFORE importing router ----
const runOrder: string[] = [];
vi.mock("../../src/agent/runner", () => ({
runAgent: vi.fn(async ({ userMessage }: { userMessage: string }) => {
// Simulate brief async work so the promise chain is exercised
await new Promise<void>((resolve) => setTimeout(resolve, 5));
runOrder.push(userMessage);
return `echo: ${userMessage}`;
}),
}));
// ---- Import after mock registration ----
import { routeMessage } from "../../src/router";
// ---- Helpers ----
/** Builds a minimal fake discord.js Message object. */
function makeMessage(channelId: string, content: string) {
const sent: string[] = [];
const channel = {
name: "test-channel",
sendTyping: vi.fn().mockResolvedValue(undefined),
send: vi.fn(async (text: string) => {
sent.push(text);
return { id: `msg-${Math.random()}` };
}),
};
return {
channelId,
content,
id: `mid-${content}`,
author: { username: "tester" },
channel,
_sent: sent,
} as unknown as import("discord.js").Message & { _sent: string[] };
}
/** Minimal fake DisclawDatabase. */
function makeDb(channelId: string) {
return {
getWorkspaceByChannelId: vi.fn((_id: string) => ({
id: 1,
workspace_path: "/fake/workspace",
})),
getConversationHistory: vi.fn(() => []),
addConversation: vi.fn(),
} as unknown as import("../../src/db/database").DisclawDatabase;
}
/** Minimal fake DisclawConfig. */
const fakeConfig = {} as import("../../src/config/loader").DisclawConfig;
// ---- Tests ----
describe("channel-queue / router integration", () => {
beforeEach(() => {
runOrder.length = 0;
});
afterEach(() => {
vi.restoreAllMocks();
});
it("processes 3 messages in the same channel in FIFO order", async () => {
const channelId = "ch-999";
const db = makeDb(channelId);
// Fire all three without awaiting — they should serialise internally
const p1 = routeMessage(makeMessage(channelId, "first"), db, fakeConfig, null);
const p2 = routeMessage(makeMessage(channelId, "second"), db, fakeConfig, null);
const p3 = routeMessage(makeMessage(channelId, "third"), db, fakeConfig, null);
await Promise.all([p1, p2, p3]);
expect(runOrder).toEqual(["first", "second", "third"]);
});
it("returns true for every handled message", async () => {
const channelId = "ch-handled";
const db = makeDb(channelId);
const results = await Promise.all([
routeMessage(makeMessage(channelId, "a"), db, fakeConfig, null),
routeMessage(makeMessage(channelId, "b"), db, fakeConfig, null),
]);
expect(results).toEqual([true, true]);
});
it("returns false when channel is the management channel", async () => {
const channelId = "ch-mgmt";
const db = makeDb(channelId);
const result = await routeMessage(
makeMessage(channelId, "hi"),
db,
fakeConfig,
channelId // same id => management channel
);
expect(result).toBe(false);
});
});

View file

@ -0,0 +1,100 @@
import { describe, it, expect, vi } from "vitest";
import { ChannelQueue } from "../../src/runtime/channel-queue";
/**
* Returns a task that resolves after `ms` milliseconds and pushes its label
* into the shared `order` array.
*/
function makeTask(order: string[], label: string, ms = 0): () => Promise<void> {
return () =>
new Promise<void>((resolve) => {
setTimeout(() => {
order.push(label);
resolve();
}, ms);
});
}
describe("ChannelQueue", () => {
it("(a) executes tasks in FIFO order within the same channel", async () => {
const queue = new ChannelQueue();
const order: string[] = [];
const p1 = queue.enqueue("ch1", makeTask(order, "first"));
const p2 = queue.enqueue("ch1", makeTask(order, "second"));
const p3 = queue.enqueue("ch1", makeTask(order, "third"));
await Promise.all([p1, p2, p3]);
expect(order).toEqual(["first", "second", "third"]);
});
it("(b) a failing task does not block the next task", async () => {
const queue = new ChannelQueue();
const order: string[] = [];
const failing: () => Promise<void> = () =>
Promise.reject(new Error("intentional failure"));
// Suppress the console.error output from ChannelQueue internals
const spy = vi.spyOn(console, "error").mockImplementation(() => {});
const p1 = queue.enqueue("ch1", failing);
const p2 = queue.enqueue("ch1", makeTask(order, "after-error"));
await Promise.all([p1, p2]);
spy.mockRestore();
expect(order).toEqual(["after-error"]);
});
it("(c) map entry is removed after all tasks complete", async () => {
const queue = new ChannelQueue();
const p1 = queue.enqueue("ch1", () => Promise.resolve());
const p2 = queue.enqueue("ch1", () => Promise.resolve());
await Promise.all([p1, p2]);
// Allow the finally-cleanup microtask to run
await Promise.resolve();
expect((queue as unknown as { queues: Map<string, unknown> }).queues.size).toBe(0);
});
it("(d) different channels run in parallel, not serialized", async () => {
const queue = new ChannelQueue();
const completionOrder: string[] = [];
// ch-slow gets a 50 ms task; ch-fast gets a 10 ms task.
// If channels were serialized (ch-slow first), ch-fast would finish after
// ch-slow. With true parallelism ch-fast finishes first.
const pSlow = queue.enqueue(
"ch-slow",
() =>
new Promise<void>((resolve) => {
setTimeout(() => {
completionOrder.push("slow");
resolve();
}, 50);
})
);
const pFast = queue.enqueue(
"ch-fast",
() =>
new Promise<void>((resolve) => {
setTimeout(() => {
completionOrder.push("fast");
resolve();
}, 10);
})
);
await Promise.all([pSlow, pFast]);
// fast must finish before slow
expect(completionOrder).toEqual(["fast", "slow"]);
});
});