talon_http/
investigate.rs

1//! Investigation jobs and the seeded alert-digest task.
2//!
3//! `POST /investigate` registers a `once` scheduled task on the
4//! investigation lane (due immediately) plus an [`Investigation`] row, then
5//! wakes the scheduler. Running it through the task scheduler instead of the
6//! chat message loop makes the job durable across restarts and lets the
7//! host correlate task id == job id when it records the outcome.
8
9use std::time::Duration;
10
11use chrono::{DateTime, Utc};
12use chrono_tz::Tz;
13use talon_core::schedule;
14use talon_core::time::{parse_iso, to_iso};
15use talon_core::types::{ContextMode, ScheduleType, ScheduledTask, TaskStatus};
16use talon_core::{Error, Result};
17use talon_db::Investigation;
18
19use crate::store::HttpStore;
20
21/// Id of the seeded 15-minute alert digest task.
22pub const DIGEST_TASK_ID: &str = "copilot-alert-digest-15m";
23/// JID the digest posts to (owned by the outbound webhook channel).
24pub const WEBHOOK_JID: &str = "webhook:copilot";
25/// Digest schedule.
26pub const DIGEST_CRON: &str = "*/15 * * * *";
27/// Investigation status written when a job is queued.
28pub const STATUS_QUEUED: &str = "queued";
29/// Investigation status while its task runs (written by the host).
30pub const STATUS_RUNNING: &str = "running";
31/// Terminal failure status; failed jobs never de-duplicate a new trigger.
32pub const STATUS_FAILED: &str = "failed";
33
34/// A validated investigation request.
35#[derive(Debug, Clone, PartialEq, Eq)]
36pub struct InvestigationRequest {
37    /// CoPilot's numeric alert id.
38    pub alert_id: u64,
39    /// Customer code (`^[A-Za-z0-9_-]{1,64}$`).
40    pub customer_code: String,
41    /// Trigger label (`webhook` by default).
42    pub triggered_by: String,
43    /// Template filename forced by the caller.
44    pub template_override: Option<String>,
45}
46
47/// `copilot-inv-<alert_id>-<millis>`
48pub fn job_id(alert_id: u64, millis: i64) -> String {
49    format!("copilot-inv-{alert_id}-{millis}")
50}
51
52/// Whether `existing` should be returned instead of starting a new job for
53/// `req` at `now`: same alert, customer and template, not failed, created
54/// within `window`. A zero window disables de-duplication. A different
55/// template override is a deliberate replay and always runs.
56pub fn is_duplicate(
57    existing: &Investigation,
58    req: &InvestigationRequest,
59    window: Duration,
60    now: DateTime<Utc>,
61) -> bool {
62    if window.is_zero()
63        || existing.status == STATUS_FAILED
64        || existing.alert_id != req.alert_id.to_string()
65        || existing.customer_code != req.customer_code
66        || existing.template != req.template_override
67    {
68        return false;
69    }
70    let Some(created) = parse_iso(&existing.created_at) else {
71        return false;
72    };
73    let Ok(window) = chrono::Duration::from_std(window) else {
74        return true;
75    };
76    created > now - window && created <= now + chrono::Duration::minutes(5)
77}
78
79/// The `once` task that runs an investigation on `chat_jid`, due now.
80pub fn investigation_task(
81    id: &str,
82    group_folder: &str,
83    chat_jid: &str,
84    prompt: String,
85    tz: Tz,
86    now: DateTime<Utc>,
87) -> Result<ScheduledTask> {
88    let local = now
89        .with_timezone(&tz)
90        .format("%Y-%m-%dT%H:%M:%S")
91        .to_string();
92    let next_run = schedule::first_run(ScheduleType::Once, &local, tz, now)?;
93    Ok(ScheduledTask {
94        id: id.to_string(),
95        group_folder: group_folder.to_string(),
96        chat_jid: chat_jid.to_string(),
97        prompt,
98        script: None,
99        schedule_type: ScheduleType::Once,
100        schedule_value: local,
101        context_mode: ContextMode::Isolated,
102        next_run: Some(next_run),
103        last_run: None,
104        last_result: None,
105        status: TaskStatus::Active,
106        created_at: to_iso(now),
107    })
108}
109
110/// The investigation row stored alongside the task.
111pub fn investigation_record(
112    id: &str,
113    req: &InvestigationRequest,
114    now: DateTime<Utc>,
115) -> Investigation {
116    let ts = to_iso(now);
117    Investigation {
118        id: id.to_string(),
119        alert_id: req.alert_id.to_string(),
120        customer_code: req.customer_code.clone(),
121        triggered_by: req.triggered_by.clone(),
122        status: STATUS_QUEUED.to_string(),
123        template: req.template_override.clone(),
124        created_at: ts.clone(),
125        updated_at: ts,
126        verdict: None,
127        error: None,
128    }
129}
130
131/// Prompt for a targeted investigation, including the structured-verdict
132/// step (every interpolated value is validated before it gets here).
133pub fn investigation_prompt(req: &InvestigationRequest, job_id: &str) -> String {
134    let alert_id = req.alert_id;
135    let customer_code = &req.customer_code;
136    let triggered_by = &req.triggered_by;
137    let override_line = req
138        .template_override
139        .as_deref()
140        .map(|t| format!("\n- template_override: {t}"))
141        .unwrap_or_default();
142    let step_override_instruction = req
143        .template_override
144        .as_deref()
145        .map(|t| {
146            format!(
147                "3a. TEMPLATE OVERRIDE — the caller selected template \"{t}\". Skip Step 2.5 template detection entirely. Read /workspace/group/prompts/{t} directly and follow it. Record selection_method=\"override\" when writing the eval JSON.\n"
148            )
149        })
150        .unwrap_or_default();
151
152    format!(
153        r#"You are running a targeted SOC investigation triggered by CoPilot. Follow the full investigation workflow from AGENT.md.
154
155Alert to investigate:
156- alert_id: {alert_id}
157- customer_code: {customer_code}
158- job_id: {job_id}
159- triggered_by: {triggered_by}{override_line}
160
161Steps:
1621. Skip Step 0 deduplication — Talahon already deduplicated this request and assigned the job id. Register the job in CoPilot and mark it running, in this order:
163   a. mcp__copilot__CreateAiAnalystJobTool(id="{job_id}", alert_id={alert_id}, customer_code="{customer_code}", triggered_by="{triggered_by}")
164   b. mcp__copilot__UpdateAiAnalystJobTool(job_id="{job_id}", status="running")
1652. Resolve the alert source and target document via MySQL before any OpenSearch query. The alert_id passed here is CoPilot's internal DB counter and is NOT a field on raw SIEM documents — searching OpenSearch for it returns nothing, and falling back to a time-window query will pick up unrelated noise (typically the agent's own container-startup events) and produce a report about the wrong event. Run exactly this query via mcp__mysql__mysql_query:
166
167     SELECT a.id, a.alert_name, a.source, a.alert_creation_time,
168            a.customer_code, ast.asset_name, ast.agent_id,
169            ast.index_name, ast.index_id
170     FROM incident_management_alert a
171     LEFT JOIN incident_management_asset ast ON ast.alert_linked = a.id
172     WHERE a.id = {alert_id} AND a.customer_code = "{customer_code}";
173
174   Treat the returned (index_name, index_id) as the authoritative pointer to the raw event for every subsequent SIEM lookup. The 'source' column ("wazuh", "office365", "misp", …) tells you which OpenSearch index family the event lives in. If the row is missing, ast.index_name is empty, or the source is unrecognised, mark the job 'failed' with a descriptive error_message rather than guessing from timestamps.
1753. Follow Step 0.5 (MemPalace prior context for the asset, customer and IOCs) and Steps 2–6 from AGENT.md exactly (fetch raw event + index mapping, detect alert type, load template, extract IOCs, threat intel, SIEM correlation).
176{step_override_instruction}3b. Complete Step 5.5 from AGENT.md: call score_alert, verify every claim against the evidence you collected, then call submit_triage_report with report_id="{job_id}". Use the severity_assessment label and the Markdown it returns for step 4a. Talahon records this structured verdict on the investigation.
1774. Write back to CoPilot — call these three tools in order. The four body fields on SubmitAiAnalystReportTool (severity_assessment, summary, report_markdown, recommended_actions) carry the entire human-readable investigation output; a call missing any of them persists a blank row that surfaces as an empty report in CoPilot. Pass all of them every time, even on trivial false positives.
178
179   a. mcp__copilot__SubmitAiAnalystReportTool — returns report_id:
180        job_id="{job_id}"
181        alert_id={alert_id}
182        customer_code="{customer_code}"
183        severity_assessment=<Critical|High|Medium|Low|Informational>
184        summary=<1-2 sentence tl;dr>
185        report_markdown=<full structured markdown report>
186        recommended_actions=<action list>
187
188   b. mcp__copilot__SubmitAiAnalystIocsTool(report_id=<from 4a>, alert_id={alert_id}, customer_code="{customer_code}", iocs=[...])
189
190   c. Only after 4a succeeded: mcp__copilot__UpdateAiAnalystJobTool(job_id="{job_id}", status="completed", template_used=<template key or null>). If 4a failed, set status="failed" with error_message instead: a job marked completed without a report is never investigated again.
191
1925. Write the eval JSON to /home/agent/.talon/evals/{alert_id}-{job_id}.json as instructed in AGENT.md Step 6 (create the directory if needed). That directory is private to this investigation lane; never write evals under /workspace/group.
1936. Send the full investigation report via send_message."#
194    )
195}
196
197/// The paused 15-minute alert digest task, seeded once per database.
198pub fn digest_task(group_folder: &str, tz: Tz, now: DateTime<Utc>) -> Result<ScheduledTask> {
199    let next_run = schedule::first_run(ScheduleType::Cron, DIGEST_CRON, tz, now)?;
200    Ok(ScheduledTask {
201        id: DIGEST_TASK_ID.to_string(),
202        group_folder: group_folder.to_string(),
203        chat_jid: WEBHOOK_JID.to_string(),
204        prompt: DIGEST_PROMPT.to_string(),
205        script: None,
206        schedule_type: ScheduleType::Cron,
207        schedule_value: DIGEST_CRON.to_string(),
208        // A fresh session per run: the webhook lane's chat history and an
209        // earlier digest's raw SIEM data have no business in this one.
210        context_mode: ContextMode::Isolated,
211        next_run: Some(next_run),
212        last_run: None,
213        last_result: None,
214        status: TaskStatus::Paused,
215        created_at: to_iso(now),
216    })
217}
218
219/// Create the digest task unless it already exists. Returns true if it was
220/// created. Never touches an existing task (an operator may have resumed or
221/// edited it).
222pub fn seed_digest_task(
223    store: &dyn HttpStore,
224    group_folder: &str,
225    tz: Tz,
226    now: DateTime<Utc>,
227) -> Result<bool> {
228    if store.get_task_by_id(DIGEST_TASK_ID)?.is_some() {
229        return Ok(false);
230    }
231    let task = digest_task(group_folder, tz, now)?;
232    store.create_task(&task)?;
233    Ok(true)
234}
235
236/// Create the task and the investigation row; on a failure to record the
237/// investigation, the task is removed again so the two never drift apart.
238pub(crate) fn persist_investigation(
239    store: &dyn HttpStore,
240    task: &ScheduledTask,
241    inv: &Investigation,
242) -> Result<()> {
243    if store.get_task_by_id(&task.id)?.is_some() {
244        return Err(Error::invalid(format!("task {} already exists", task.id)));
245    }
246    store.create_task(task)?;
247    if let Err(e) = store.upsert_investigation(inv) {
248        if let Err(del) = store.delete_task(&task.id) {
249            tracing::error!(task = %task.id, err = %del, "failed to roll back investigation task");
250        }
251        return Err(e);
252    }
253    Ok(())
254}
255
256/// Prompt of the scheduled SOC monitor (digest) task.
257pub const DIGEST_PROMPT: &str = r#"You are running as a scheduled SOC monitor. Follow these steps exactly for each alert.
258
2591. Query MySQL for OPEN alerts of the last 24 hours that have no AI
260   investigation yet. A completed job, or a pending/running one younger
261   than 2 hours (an older one is a dead run), counts as an investigation;
262   failed jobs do not:
263
264   SELECT a.id, a.alert_name, a.source, a.alert_creation_time,
265          a.customer_code, ast.asset_name, ast.agent_id,
266          ast.index_name, ast.index_id
267   FROM incident_management_alert a
268   JOIN incident_management_asset ast ON ast.alert_linked = a.id
269   WHERE a.status = 'OPEN'
270     AND a.alert_creation_time >= DATE_SUB(NOW(), INTERVAL 24 HOUR)
271     AND ast.index_name != ''
272     AND ast.index_id != ''
273     AND NOT EXISTS (
274       SELECT 1 FROM ai_analyst_job j
275       WHERE j.alert_id = a.id
276         AND (j.status = 'completed'
277              OR (j.status IN ('pending', 'running')
278                  AND j.created_at >= DATE_SUB(NOW(), INTERVAL 2 HOUR)))
279     )
280   ORDER BY a.alert_creation_time DESC
281   LIMIT 20;
282
2832. If no rows are returned, stop here. Do not send any message.
284
2853. For each alert:
286
287   a. DEDUPLICATION — call ListAiAnalystJobsByAlertTool(alert_id=<id>)
288      right before you start (a real-time investigation may have begun
289      since the query). Skip this alert if a job is "completed", or
290      "pending"/"running" and created less than 2 hours ago.
291
292   b. REGISTER JOB — call CreateAiAnalystJobTool:
293        id="copilot-inv-<alert_id>-<unix_timestamp>"
294        alert_id=<id>, customer_code=<code>, triggered_by="scheduled"
295      Then immediately call UpdateAiAnalystJobTool(job_id=<id>, status="running").
296
297   c. FETCH INDEX MAPPING + RAW EVENT in parallel:
298      - get_document(index_name, index_id) → full raw alert event
299      - get_index(index_name) → field mapping (confirms field names and types)
300      Use the mapping before writing any OpenSearch query.
301
302   d. DETECT ALERT TYPE from the raw event:
303      - List available templates: ls /workspace/group/prompts/
304      - Compare the template filenames with the alert summary
305        (rule_description, rule_groups, data_win_system_eventID, agent_name)
306        and pick the best matching filename, or none if nothing fits
307        (selection_method="hints")
308      - Fallback: match rule_groups values or data_win_system_eventID
309        against the available filenames (selection_method="fallback")
310
311      NOTE: Graylog flattens all fields with underscores — use
312      rule_groups, rule_description, data_win_system_eventID, agent_name.
313      Never use dot notation for raw document fields.
314
315   e. LOAD TEMPLATE — Read /workspace/group/prompts/<filename>.
316      If found, follow its steps substituting:
317        {{ alert }} → raw OpenSearch event JSON
318        {{ event_id }} → numeric event ID
319        {{ pipeline | default('wazuh') }} → wazuh
320        {{ virustotal_results }} → VT results after threat intel
321      If no template, use the default steps below.
322
323   f. DEFAULT INVESTIGATION (when no template matches):
324      - Extract IOCs: IPs (data_win_eventdata_destinationIp, data_srcip, data_dstip),
325        domains (data_win_eventdata_queryName, data_win_eventdata_destinationHostname),
326        hashes (data_win_eventdata_hashes), processes (data_win_eventdata_image,
327        data_win_eventdata_parentImage), commands (data_win_eventdata_commandLine)
328      - For each external IP/domain: WebSearch "<value>" site:virustotal.com
329      - For each SHA256 hash: WebFetch https://www.virustotal.com/gui/file/<hash>
330      - Check rule_level, rule_description, rule_mitre_tactic
331
332   g. WRITE BACK TO COPILOT — call in order:
333      1. SubmitAiAnalystReportTool(
334           job_id=<id>, alert_id=<id>, customer_code=<code>,
335           severity_assessment=<Critical|High|Medium|Low|Informational>,
336           summary=<1-2 sentence tl;dr>,
337           report_markdown=<full markdown report>,
338           recommended_actions=<action list>)
339         → note the returned report_id
340      2. SubmitAiAnalystIocsTool(
341           report_id=<from step 1>, alert_id=<id>, customer_code=<code>,
342           iocs=[{ ioc_value, ioc_type, vt_verdict, vt_score, details }, ...])
343         ioc_type: ip|domain|hash|process|url|user|command
344         vt_verdict: malicious|suspicious|clean|unknown
345      3. Only after step 1 succeeded: UpdateAiAnalystJobTool(job_id=<id>,
346           status="completed", template_used=<alert_type or null>).
347         If step 1 failed: status="failed" with error_message.
348
349   h. SEND REPORT via send_message:
350      🔍 **SOC Investigation** — <alert_name>
351      Customer: <customer_code> | Asset: <asset_name> | Created: <alert_creation_time>
352      **Severity**: <assessment> | **Template**: <alert_type or "default">
353
354      **Alert Summary**: rule description, MITRE tactic/technique
355      **IOC Analysis**: table of IOCs with type and VT verdict
356      **Severity Assessment**: reasoning
357      **Recommended Actions**: specific actionable steps
358
359      On any error during write-back, call UpdateAiAnalystJobTool with
360      status="failed" and error_message=<exception details>."#;
361
362#[cfg(test)]
363mod tests {
364    use super::*;
365
366    fn req(template: Option<&str>) -> InvestigationRequest {
367        InvestigationRequest {
368            alert_id: 1234,
369            customer_code: "acme".into(),
370            triggered_by: "webhook".into(),
371            template_override: template.map(str::to_string),
372        }
373    }
374
375    fn now() -> DateTime<Utc> {
376        parse_iso("2026-10-14T12:00:00.000Z").unwrap()
377    }
378
379    #[test]
380    fn job_ids_follow_the_ts_format() {
381        assert_eq!(
382            job_id(1234, 1714000000000),
383            "copilot-inv-1234-1714000000000"
384        );
385    }
386
387    #[test]
388    fn prompt_carries_the_job_and_alert() {
389        let p = investigation_prompt(&req(None), "copilot-inv-1234-1");
390        assert!(p.contains("- alert_id: 1234\n- customer_code: acme\n- job_id: copilot-inv-1234-1\n- triggered_by: webhook\n\nSteps:"));
391        assert!(p.contains(r#"WHERE a.id = 1234 AND a.customer_code = "acme";"#));
392        assert!(p.contains(
393            "5. Write the eval JSON to /home/agent/.talon/evals/1234-copilot-inv-1234-1.json as instructed"
394        ));
395        assert!(!p.contains("/workspace/group/evals"));
396        assert!(!p.contains("TEMPLATE OVERRIDE"));
397        assert!(p.contains("SIEM correlation).\n3b. Complete Step 5.5"));
398        assert!(p.contains(r#"submit_triage_report with report_id="copilot-inv-1234-1""#));
399        assert!(p.contains("for step 4a. Talahon records this structured verdict on the investigation.\n4. Write back"));
400        // MemPalace lessons are recalled on the real-time path too.
401        assert!(p.contains("3. Follow Step 0.5 (MemPalace"));
402        // The job is completed only after its report exists.
403        let report = p
404            .find("a. mcp__copilot__SubmitAiAnalystReportTool")
405            .unwrap();
406        let completed = p
407            .find(r#"UpdateAiAnalystJobTool(job_id="copilot-inv-1234-1", status="completed""#)
408            .unwrap();
409        assert!(report < completed);
410        // The agent registers the job itself, carrying the real trigger, before
411        // marking it running (CoPilot's auto-create would record "webhook").
412        let create = p
413            .find(r#"CreateAiAnalystJobTool(id="copilot-inv-1234-1", alert_id=1234, customer_code="acme", triggered_by="webhook")"#)
414            .unwrap();
415        let running = p
416            .find(r#"UpdateAiAnalystJobTool(job_id="copilot-inv-1234-1", status="running")"#)
417            .unwrap();
418        assert!(create < running);
419    }
420
421    #[test]
422    fn prompt_includes_the_template_override() {
423        let p = investigation_prompt(&req(Some("sysmon_event_1.txt")), "j");
424        assert!(p.contains("- triggered_by: webhook\n- template_override: sysmon_event_1.txt\n"));
425        assert!(p.contains(
426            "3a. TEMPLATE OVERRIDE — the caller selected template \"sysmon_event_1.txt\"."
427        ));
428        assert!(p.contains("Read /workspace/group/prompts/sysmon_event_1.txt directly"));
429        assert!(p.contains(
430            "selection_method=\"override\" when writing the eval JSON.\n3b. Complete Step 5.5"
431        ));
432    }
433
434    #[test]
435    fn once_task_is_due_now_on_the_investigation_lane() {
436        let t = investigation_task(
437            "copilot-inv-1-5",
438            "copilot",
439            "http:copilot:investigate",
440            "p".into(),
441            chrono_tz::Europe::Berlin,
442            now(),
443        )
444        .unwrap();
445        assert_eq!(t.schedule_type, ScheduleType::Once);
446        assert_eq!(t.schedule_value, "2026-10-14T14:00:00");
447        assert_eq!(t.next_run.as_deref(), Some("2026-10-14T12:00:00.000Z"));
448        assert_eq!(t.context_mode, ContextMode::Isolated);
449        assert_eq!(t.status, TaskStatus::Active);
450        assert_eq!(t.chat_jid, "http:copilot:investigate");
451        schedule::validate(
452            t.schedule_type,
453            &t.schedule_value,
454            chrono_tz::Europe::Berlin,
455        )
456        .unwrap();
457    }
458
459    #[test]
460    fn dedup_rules() {
461        let window = Duration::from_secs(600);
462        let mut existing = investigation_record("j1", &req(None), now());
463        let later = now() + chrono::Duration::minutes(5);
464        assert!(is_duplicate(&existing, &req(None), window, later));
465        // Outside the window.
466        assert!(!is_duplicate(
467            &existing,
468            &req(None),
469            window,
470            now() + chrono::Duration::minutes(11)
471        ));
472        // Disabled.
473        assert!(!is_duplicate(&existing, &req(None), Duration::ZERO, later));
474        // A different template is a replay.
475        assert!(!is_duplicate(&existing, &req(Some("x.txt")), window, later));
476        // Different customer.
477        let mut other = req(None);
478        other.customer_code = "beta".into();
479        assert!(!is_duplicate(&existing, &other, window, later));
480        // Failed jobs never block a retry.
481        existing.status = STATUS_FAILED.into();
482        assert!(!is_duplicate(&existing, &req(None), window, later));
483        // Completed jobs inside the window are returned.
484        existing.status = "completed".into();
485        assert!(is_duplicate(&existing, &req(None), window, later));
486        // Garbage timestamps never dedup.
487        existing.created_at = "yesterday".into();
488        assert!(!is_duplicate(&existing, &req(None), window, later));
489    }
490
491    #[test]
492    fn digest_task_is_paused_and_scheduled() {
493        let t = digest_task("copilot", chrono_tz::UTC, now()).unwrap();
494        assert_eq!(t.id, DIGEST_TASK_ID);
495        assert_eq!(t.status, TaskStatus::Paused);
496        assert_eq!(t.chat_jid, WEBHOOK_JID);
497        assert_eq!(t.context_mode, ContextMode::Isolated);
498        assert_eq!(t.schedule_value, "*/15 * * * *");
499        assert_eq!(t.next_run.as_deref(), Some("2026-10-14T12:15:00.000Z"));
500        assert!(
501            t.prompt
502                .starts_with("You are running as a scheduled SOC monitor.")
503        );
504        assert!(t.prompt.contains("{{ alert }} → raw OpenSearch event JSON"));
505        // Alerts already being investigated are left out in SQL, and a late
506        // run still finds alerts older than one 15-minute window.
507        assert!(t.prompt.contains("NOT EXISTS ("));
508        assert!(t.prompt.contains("INTERVAL 24 HOUR"));
509        assert!(!t.prompt.contains("INTERVAL 15 MINUTE"));
510        let report = t.prompt.find("1. SubmitAiAnalystReportTool(").unwrap();
511        let completed = t
512            .prompt
513            .find("UpdateAiAnalystJobTool(job_id=<id>,\n           status=\"completed\"")
514            .unwrap();
515        assert!(report < completed);
516        assert!(t.prompt.ends_with("error_message=<exception details>."));
517    }
518}
talon_http/investigate.rs

talon_http/
investigate.rs
