From faf3193d2ff02a491983e941e1ed61d83f0c2ad9 Mon Sep 17 00:00:00 2001 From: Johnny Huynh <27847622+johnnyhuy@users.noreply.github.com> Date: Sun, 8 Nov 2026 23:22:38 +0900 Subject: [PATCH] feat(rules): honour operator rules on the Slack submit path --- backend/app/rag.py | 20 +++++++++++++++- backend/tests/test_rules.py | 48 +++++++++++++++++++++++++++++++++++++ runtime/agent.test.mjs | 20 ++++++++++++++++ runtime/jobs.mjs | 7 +++++- runtime/server.mjs | 1 + 5 files changed, 94 insertions(+), 2 deletions(-) diff --git a/backend/app/rag.py b/backend/app/rag.py index 404ec8c..eda94fc 100644 --- a/backend/app/rag.py +++ b/backend/app/rag.py @@ -141,7 +141,25 @@ def generate_response(self, question: str, channel_id: str = None, )) def submit_slack(self, question, channel_id, conversation_id, request_id, thread_ts): - """Durably enqueue generation and delivery before returning to Slack Bolt.""" + """Durably enqueue generation and delivery before returning to Slack Bolt. + + Operator rules run first, as in generate_response. A pinned answer + skips retrieval and model generation, but still goes through the + Pi job outbox so Slack delivery stays durable and idempotent. + """ + matched = self.match_operator_rule(question) + if matched is not None: + logger.info("Operator rule %s answered a Slack question", matched.rule.id) + identifier = request_id or agent_runtime.request_id() + return agent_runtime.submit({ + "request_id": identifier, + "conversation_id": conversation_id or f"request:{identifier}", + "question": question, + "context": "", + "sources": [{"source": "operator_rule", "rule_id": matched.rule.id}], + "operator_answer": matched.rule.response, + "slack": {"channel": channel_id, "thread_ts": thread_ts}, + }) return agent_runtime.submit(self._agent_payload( question, channel_id, conversation_id, request_id, slack={"channel": channel_id, "thread_ts": thread_ts}, diff --git a/backend/tests/test_rules.py b/backend/tests/test_rules.py index 56ef97e..5b7ff30 100644 --- a/backend/tests/test_rules.py +++ b/backend/tests/test_rules.py @@ -161,3 +161,51 @@ def test_unmatched_question_still_uses_generation(self, monkeypatch): monkeypatch.setattr(rag_module.agent_runtime, "generate", lambda _: expected) assert rag_module.rag_engine.generate_response("What is RAG?") == expected + + +class TestSlackShortCircuit: + def _install_engine(self, monkeypatch, eng): + monkeypatch.setattr(rag_module.rules, "rule_engine", eng) + + def test_matched_slack_question_skips_retrieval_and_carries_pinned_answer(self, monkeypatch): + self._install_engine(monkeypatch, engine(rule("vpn", ["reset vpn"], response="Use the portal."))) + search = Mock(side_effect=AssertionError("retrieval must not run")) + monkeypatch.setattr(rag_module.vector_store, "similarity_search", search) + submitted = {} + def fake_submit(payload): + submitted["payload"] = payload + return {"id": "job-1", "status": "pending"} + monkeypatch.setattr(rag_module.agent_runtime, "submit", fake_submit) + + result = rag_module.rag_engine.submit_slack( + "How do I reset VPN access?", "C1", "team:thread:C1:1.2", "event-1", "1.2" + ) + + assert result == {"id": "job-1", "status": "pending"} + payload = submitted["payload"] + assert payload["operator_answer"] == "Use the portal." + assert payload["sources"] == [{"source": "operator_rule", "rule_id": "vpn"}] + assert payload["context"] == "" + assert payload["question"] == "How do I reset VPN access?" + assert payload["request_id"] == "event-1" + assert payload["conversation_id"] == "team:thread:C1:1.2" + assert payload["slack"] == {"channel": "C1", "thread_ts": "1.2"} + + def test_unmatched_slack_question_uses_retrieval_without_pinned_answer(self, monkeypatch): + self._install_engine(monkeypatch, engine(rule("vpn", ["reset vpn"]))) + monkeypatch.setattr( + rag_module.vector_store, "similarity_search", + lambda _: [{"content": "Doc", "metadata": {"source": "slack", "channel_id": "C1"}}], + ) + submitted = {} + def fake_submit(payload): + submitted["payload"] = payload + return {"id": "job-2", "status": "pending"} + monkeypatch.setattr(rag_module.agent_runtime, "submit", fake_submit) + + rag_module.rag_engine.submit_slack("What is RAG?", "C1", "team:thread:C1:1.2", "event-2", "1.2") + + payload = submitted["payload"] + assert "operator_answer" not in payload + assert "Doc" in payload["context"] + assert payload["slack"] == {"channel": "C1", "thread_ts": "1.2"} diff --git a/runtime/agent.test.mjs b/runtime/agent.test.mjs index adc1571..b5e7f3b 100644 --- a/runtime/agent.test.mjs +++ b/runtime/agent.test.mjs @@ -81,6 +81,26 @@ test('a process killed during generation resumes the admitted Pi submission', { }); +test('pinned operator answer is delivered without calling the model', async () => { + const jobs = new Jobs(':memory:'); + try { + const job = jobs.submit({ ...payload, slack: { channel: 'C1', thread_ts: '1.2' }, + operator_answer: 'Use the portal.', sources: [{ source: 'operator_rule', rule_id: 'vpn' }] }); + assert.throws(() => jobs.submit({ ...payload, slack: { channel: 'C1', thread_ts: '1.2' }, + operator_answer: 'Different pinned answer.', sources: [{ source: 'operator_rule', rule_id: 'vpn' }] }), /reused/); + let calls = 0; + let delivered; + await processJob(jobs, job, async () => { calls++; throw new Error('model must not run'); }, + async (_id, _target, text) => { delivered = text; }); + assert.equal(calls, 0); + assert.equal(delivered, 'Use the portal.'); + assert.equal(jobs.get(job.id).status, 'done'); + assert.deepEqual(jobs.get(job.id).result, { + answer: 'Use the portal.', sources: [{ source: 'operator_rule', rule_id: 'vpn' }], confidence: 'high', + }); + } finally { jobs.close(); } +}); + test('model failure sends a persisted Slack failure notification', async () => { const jobs = new Jobs(':memory:'); try { diff --git a/runtime/jobs.mjs b/runtime/jobs.mjs index 715e9eb..2b312f6 100644 --- a/runtime/jobs.mjs +++ b/runtime/jobs.mjs @@ -17,6 +17,7 @@ export class Jobs { if (previous) { const old = previous.payload; if (old.question !== payload.question || old.conversation_id !== payload.conversation_id || + (old.operator_answer ?? null) !== (payload.operator_answer ?? null) || JSON.stringify(old.slack) !== JSON.stringify(payload.slack)) { throw Object.assign(new Error('Request ID reused for different input'), { statusCode: 409 }); } @@ -60,7 +61,11 @@ export class Jobs { export async function processJob(jobs, job, answer, deliver) { if (!job.result) { try { - job.result = await answer(job.payload); + // A pinned operator answer needs no model call. It still flows + // through generated/delivered so Slack delivery retries persist. + job.result = typeof job.payload.operator_answer === 'string' && job.payload.operator_answer + ? { answer: job.payload.operator_answer, sources: job.payload.sources ?? [], confidence: 'high' } + : await answer(job.payload); jobs.generated(job.id, job.result, Boolean(job.payload.slack)); } catch (error) { if (!job.payload.slack) { diff --git a/runtime/server.mjs b/runtime/server.mjs index 834d593..d842a97 100644 --- a/runtime/server.mjs +++ b/runtime/server.mjs @@ -74,6 +74,7 @@ const server = createServer(async (request, response) => { try { body = JSON.parse(raw); } catch { return send(400, { error: 'Invalid JSON' }); } if (!body || !['request_id','conversation_id','question','context'].every(key => typeof body[key] === 'string') || !body.request_id || !body.conversation_id || !body.question || !Array.isArray(body.sources) || + (body.operator_answer != null && typeof body.operator_answer !== 'string') || (body.slack && (typeof body.slack.channel !== 'string' || (body.slack.thread_ts != null && typeof body.slack.thread_ts !== 'string')))) { return send(400, { error: 'Invalid job' });