Skip to content

feat: route discuss messages through task queue for tracing #19

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jacksoncage merged 2 commits into from
Mar 2, 2026
Merged

feat: route discuss messages through task queue for tracing #19

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a69431f
feat: route discuss messages through task queue for tracing
jacksoncage Mar 1, 2026
a3605f7
fix: add rowid tiebreaker to task queue ordering
jacksoncage Mar 2, 2026
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
129 changes: 35 additions & 94 deletions src/handlers.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -171,23 +171,15 @@ describe("getPersona (tested via createMessageHandler discuss path)", () => {
}),
});

// In strict mode, discuss prompt should contain OVE_PERSONA but NOT the addendum
let capturedPrompt = "";
const capturingRunner: AgentRunner = {
name: "capture",
run: async (prompt: string) => {
capturedPrompt = prompt;
return { success: true, output: "response", durationMs: 50 };
},
};
deps.getRunner = () => capturingRunner;

const handler = createMessageHandler(deps);
const msg = makeMessage("discuss testing strategies");
await handler(msg);

expect(capturedPrompt).toContain("grumpy");
expect(capturedPrompt).not.toContain("IMPORTANT MODE OVERRIDE");
const tasks = deps.queue.listByUser("slack:U123", 1);
expect(tasks.length).toBe(1);
expect(tasks[0].taskType).toBe("discuss");
expect(tasks[0].prompt).toContain("grumpy");
expect(tasks[0].prompt).not.toContain("IMPORTANT MODE OVERRIDE");
});

it("appends ASSISTANT_ADDENDUM in assistant mode", async () => {
Expand All @@ -198,26 +190,18 @@ describe("getPersona (tested via createMessageHandler discuss path)", () => {
}),
});

let capturedPrompt = "";
const capturingRunner: AgentRunner = {
name: "capture",
run: async (prompt: string) => {
capturedPrompt = prompt;
return { success: true, output: "response", durationMs: 50 };
},
};
deps.getRunner = () => capturingRunner;

// Set assistant mode before sending the discuss message
deps.sessions.setMode("slack:U123", "assistant");

const handler = createMessageHandler(deps);
const msg = makeMessage("discuss testing strategies");
await handler(msg);

expect(capturedPrompt).toContain("grumpy");
expect(capturedPrompt).toContain("IMPORTANT MODE OVERRIDE");
expect(capturedPrompt).toContain("general-purpose assistant");
const tasks = deps.queue.listByUser("slack:U123", 1);
expect(tasks.length).toBe(1);
expect(tasks[0].prompt).toContain("grumpy");
expect(tasks[0].prompt).toContain("IMPORTANT MODE OVERRIDE");
expect(tasks[0].prompt).toContain("general-purpose assistant");
});
});

Expand Down Expand Up @@ -290,59 +274,39 @@ describe("createEventHandler reads mode for event.userId", () => {
});

describe("integration: set mode then send message verifies prompt contains addendum", () => {
it("set mode → send discuss → prompt sent to runner has ASSISTANT_ADDENDUM", async () => {
it("set mode → send discuss → enqueued prompt has ASSISTANT_ADDENDUM", async () => {
const deps = makeDeps({
config: makeConfig({
repos: {},
users: { "slack:U123": { name: "testuser", repos: [] } },
}),
});

let capturedPrompt = "";
const capturingRunner: AgentRunner = {
name: "capture",
run: async (prompt: string) => {
capturedPrompt = prompt;
return { success: true, output: "Here's my help!", durationMs: 50 };
},
};
deps.getRunner = () => capturingRunner;

const handler = createMessageHandler(deps);

// Step 1: set assistant mode
const modeMsg = makeMessage("mode assistant");
await handler(modeMsg);
expect(deps.sessions.getMode("slack:U123")).toBe("assistant");

// Step 2: send a discuss-type message (routes to handleDiscuss which calls runner directly)
// Step 2: send a discuss-type message (enqueued as discuss task)
const chatMsg = makeMessage("discuss best Italian restaurants nearby");
await handler(chatMsg);

// The prompt sent to the runner should contain the addendum
expect(capturedPrompt).toContain("IMPORTANT MODE OVERRIDE");
expect(capturedPrompt).toContain("willing to help with ANY request");
expect(capturedPrompt).toContain("grumpy");
const tasks = deps.queue.listByUser("slack:U123", 1);
expect(tasks[0].prompt).toContain("IMPORTANT MODE OVERRIDE");
expect(tasks[0].prompt).toContain("willing to help with ANY request");
expect(tasks[0].prompt).toContain("grumpy");
});

it("set strict mode → send discuss → prompt has NO addendum", async () => {
it("set strict mode → send discuss → enqueued prompt has NO addendum", async () => {
const deps = makeDeps({
config: makeConfig({
repos: {},
users: { "slack:U123": { name: "testuser", repos: [] } },
}),
});

let capturedPrompt = "";
const capturingRunner: AgentRunner = {
name: "capture",
run: async (prompt: string) => {
capturedPrompt = prompt;
return { success: true, output: "I only do code.", durationMs: 50 };
},
};
deps.getRunner = () => capturingRunner;

const handler = createMessageHandler(deps);

// Step 1: explicitly set strict mode (should be default, but be explicit)
Expand All @@ -354,8 +318,9 @@ describe("integration: set mode then send message verifies prompt contains adden
const chatMsg = makeMessage("discuss architecture patterns");
await handler(chatMsg);

expect(capturedPrompt).not.toContain("IMPORTANT MODE OVERRIDE");
expect(capturedPrompt).toContain("grumpy");
const tasks = deps.queue.listByUser("slack:U123", 1);
expect(tasks[0].prompt).not.toContain("IMPORTANT MODE OVERRIDE");
expect(tasks[0].prompt).toContain("grumpy");
});

it("task enqueue for repo-bound message includes addendum in assistant mode", async () => {
Expand Down Expand Up @@ -386,22 +351,13 @@ describe("mode toggle round-trip", () => {
}),
});

const prompts: string[] = [];
const capturingRunner: AgentRunner = {
name: "capture",
run: async (prompt: string) => {
prompts.push(prompt);
return { success: true, output: "ok", durationMs: 50 };
},
};
deps.getRunner = () => capturingRunner;

const handler = createMessageHandler(deps);

// Start in strict → discuss → no addendum
const msg1 = makeMessage("discuss testing");
await handler(msg1);
expect(prompts[0]).not.toContain("IMPORTANT MODE OVERRIDE");
const tasks1 = deps.queue.listByUser("slack:U123", 10);
expect(tasks1[0].prompt).not.toContain("IMPORTANT MODE OVERRIDE");

// Toggle to assistant
const modeMsg = makeMessage("mode assistant");
Expand All @@ -410,7 +366,8 @@ describe("mode toggle round-trip", () => {
// Discuss again → addendum present
const msg2 = makeMessage("discuss testing again");
await handler(msg2);
expect(prompts[1]).toContain("IMPORTANT MODE OVERRIDE");
const tasks2 = deps.queue.listByUser("slack:U123", 10);
expect(tasks2[0].prompt).toContain("IMPORTANT MODE OVERRIDE");

// Toggle back to strict
const strictMsg = makeMessage("mode strict");
Expand All @@ -419,7 +376,8 @@ describe("mode toggle round-trip", () => {
// Discuss again → no addendum
const msg3 = makeMessage("discuss testing one more time");
await handler(msg3);
expect(prompts[2]).not.toContain("IMPORTANT MODE OVERRIDE");
const tasks3 = deps.queue.listByUser("slack:U123", 10);
expect(tasks3[0].prompt).not.toContain("IMPORTANT MODE OVERRIDE");
});
});

Expand Down Expand Up @@ -474,16 +432,6 @@ describe("clear command resets mode to strict", () => {
}),
});

let capturedPrompt = "";
const capturingRunner: AgentRunner = {
name: "capture",
run: async (prompt: string) => {
capturedPrompt = prompt;
return { success: true, output: "response", durationMs: 50 };
},
};
deps.getRunner = () => capturingRunner;

const handler = createMessageHandler(deps);

// Set assistant mode
Expand All @@ -499,8 +447,9 @@ describe("clear command resets mode to strict", () => {
const chatMsg = makeMessage("discuss something fun");
await handler(chatMsg);

expect(capturedPrompt).toContain("grumpy");
expect(capturedPrompt).not.toContain("IMPORTANT MODE OVERRIDE");
const tasks = deps.queue.listByUser("slack:U123", 1);
expect(tasks[0].prompt).toContain("grumpy");
expect(tasks[0].prompt).not.toContain("IMPORTANT MODE OVERRIDE");
});
});

Expand All @@ -516,16 +465,6 @@ describe("multi-user mode isolation", () => {
}),
});

const prompts: { user: string; prompt: string }[] = [];
const capturingRunner: AgentRunner = {
name: "capture",
run: async (prompt: string) => {
prompts.push({ user: "unknown", prompt });
return { success: true, output: "response", durationMs: 50 };
},
};
deps.getRunner = () => capturingRunner;

const handler = createMessageHandler(deps);

// User A sets assistant mode
Expand All @@ -540,14 +479,16 @@ describe("multi-user mode isolation", () => {
const msgA = makeMessage("discuss best pizza places", "slack:U123");
await handler(msgA);

expect(prompts[0].prompt).toContain("IMPORTANT MODE OVERRIDE");
expect(prompts[0].prompt).toContain("grumpy");
const tasksA = deps.queue.listByUser("slack:U123", 1);
expect(tasksA[0].prompt).toContain("IMPORTANT MODE OVERRIDE");
expect(tasksA[0].prompt).toContain("grumpy");

// User B sends discuss message — should NOT have ASSISTANT_ADDENDUM
const msgB = makeMessage("discuss architecture patterns", "slack:U456");
await handler(msgB);

expect(prompts[1].prompt).not.toContain("IMPORTANT MODE OVERRIDE");
expect(prompts[1].prompt).toContain("grumpy");
const tasksB = deps.queue.listByUser("slack:U456", 1);
expect(tasksB[0].prompt).not.toContain("IMPORTANT MODE OVERRIDE");
expect(tasksB[0].prompt).toContain("grumpy");
});
});
33 changes: 9 additions & 24 deletions src/handlers.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,3 @@
import { mkdirSync } from "node:fs";
import { parseMessage, buildContextualPrompt } from "./router";
import type { ParsedMessage } from "./router";
import { isAuthorized, getUserRepos, addRepo, addUser } from "./config";
Expand Down Expand Up @@ -382,30 +381,16 @@ async function handleSchedule(msg: IncomingMessage, parsedRepo: string | undefin
async function handleDiscuss(msg: IncomingMessage, parsed: ParsedMessage, history: { role: string; content: string }[], deps: HandlerDeps) {
const prompt = buildContextualPrompt(parsed, history, getPersona(msg.userId, deps));

await msg.updateStatus("Thinking...");

try {
mkdirSync(deps.config.reposDir, { recursive: true });
const discussRunner = deps.getRunner(deps.config.runner?.name);
const result = await discussRunner.run(
prompt,
deps.config.reposDir,
{ maxTurns: 5 },
(event) => {
if (event.kind === "text") {
msg.updateStatus(event.text.slice(0, 200));
}
}
);
const taskId = deps.queue.enqueue({
userId: msg.userId,
repo: `_discuss:${crypto.randomUUID().slice(0, 8)}`,
prompt,
taskType: "discuss",
priority: parsed.priority,
});

const parts = splitAndReply(result.output, msg.platform);
for (const part of parts) {
await msg.reply(part);
}
deps.sessions.addMessage(msg.userId, "assistant", result.output.slice(0, 500));
} catch (err) {
await msg.reply(`Discussion error: ${String(err).slice(0, 500)}`);
}
deps.pendingReplies.set(taskId, msg);
logger.info("task enqueued", { taskId, type: "discuss" });
}

async function handleCreateProject(msg: IncomingMessage, parsed: ParsedMessage, history: { role: string; content: string }[], deps: HandlerDeps) {
Expand Down
8 changes: 4 additions & 4 deletions src/queue.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,7 +80,7 @@ export class TaskQueue {
`SELECT * FROM tasks
WHERE status = 'pending'
AND repo NOT IN (SELECT repo FROM tasks WHERE status = 'running')
ORDER BY priority DESC, created_at ASC
ORDER BY priority DESC, created_at ASC, rowid ASC
LIMIT 1`
)
.get() as TaskRow;
Expand Down Expand Up @@ -115,7 +115,7 @@ export class TaskQueue {
listByUser(userId: string, limit: number = 10): Task[] {
const rows = this.db
.query(
`SELECT * FROM tasks WHERE user_id = ? ORDER BY priority DESC, created_at DESC LIMIT ?`
`SELECT * FROM tasks WHERE user_id = ? ORDER BY priority DESC, created_at DESC, rowid DESC LIMIT ?`
)
.all(userId, limit) as TaskRow[];
return rows.map((r) => this.rowToTask(r));
Expand Down Expand Up @@ -206,7 +206,7 @@ export class TaskQueue {
listActive(limit: number = 20): Task[] {
const rows = this.db
.query(
`SELECT * FROM tasks WHERE status IN ('running', 'pending') ORDER BY priority DESC, created_at ASC LIMIT ?`
`SELECT * FROM tasks WHERE status IN ('running', 'pending') ORDER BY priority DESC, created_at ASC, rowid ASC LIMIT ?`
)
.all(limit) as TaskRow[];
return rows.map((r) => this.rowToTask(r));
Expand All @@ -226,7 +226,7 @@ export class TaskQueue {
sql += ` WHERE status = ?`;
params.push(status);
}
sql += ` ORDER BY priority DESC, created_at DESC LIMIT ?`;
sql += ` ORDER BY priority DESC, created_at DESC, rowid DESC LIMIT ?`;
params.push(limit);
const rows = this.db.query(sql).all(...params) as TaskRow[];
return rows.map((r) => this.rowToTask(r));
Expand Down
14 changes: 10 additions & 4 deletions src/worker.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,9 +63,11 @@ async function replyWithFallback(

async function processTask(task: Task, deps: WorkerDeps) {
const isCreateProject = task.taskType === "create-project";
const repoInfo = isCreateProject ? null : deps.getRepoInfo(task.repo);
const isDiscuss = task.taskType === "discuss";
const skipRepoSetup = isCreateProject || isDiscuss;
const repoInfo = skipRepoSetup ? null : deps.getRepoInfo(task.repo);

if (!isCreateProject && !repoInfo) {
if (!skipRepoSetup && !repoInfo) {
deps.queue.fail(task.id, `Unknown repo: ${task.repo}`);
return;
}
Expand All @@ -84,7 +86,9 @@ async function processTask(task: Task, deps: WorkerDeps) {

let workDir: string;

if (isCreateProject) {
if (isDiscuss) {
workDir = deps.config.reposDir;
} else if (isCreateProject) {
workDir = join(deps.config.reposDir, task.repo);
await Bun.write(join(workDir, ".gitkeep"), "");
} else {
Expand All @@ -108,6 +112,8 @@ async function processTask(task: Task, deps: WorkerDeps) {
const taskRunner = deps.getRunnerForRepo(task.repo);
const maxTurns = task.taskType === "cron"
? Math.max(deps.config.claude.maxTurns, 100)
: isDiscuss
? 5
: deps.config.claude.maxTurns;
const runOpts = deps.getRunnerOptsForRepo(task.repo, {
maxTurns,
Expand Down Expand Up @@ -179,7 +185,7 @@ async function processTask(task: Task, deps: WorkerDeps) {
result.success ? undefined : result.output.slice(0, 2000),
);
} finally {
if (!isCreateProject) {
if (!skipRepoSetup) {
await deps.repos.removeWorktree(task.repo, task.id).catch(() => {});
}
}
Expand Down