feat(sessions): index workflow runs and agents (layer 2 of session intelligence) (#284)

* feat(sessions): index workflow runs and their agents

Layer 2 of session intelligence (builds on #281's git spans). Indexes
Claude Code workflow runs + agents into the session store, attached to
the parent thread and to branches/worktrees/commits. Adds the
tracedecay_workflows tool (list by thread or git ref, show a run, drill
an agent), workflow_run/agent filters on message_search, and a
'sessions unfinished' CLI. Ingest is idempotent + incremental. Tool
count 100 -> 101. Verified against ~/.claude: 11 runs / 79 agents.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* test: set directory mtime cross-platform in workflow_ingest tests

The set_mtime helper opened files read-only and called set_times, which
works on Unix but panics on Windows (adjusting a directory's timestamps
needs backup-semantics access). Switch to filetime::set_file_mtime
(already a transitive dep; added as an explicit dev-dependency).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* chore: deslop workflow indexing comments

* refactor(workflow): dedup helpers and hoist loop-invariant resolution

/simplify cleanup (behavior-preserving), converged on by 4 reviewers:
- delete WorkflowStatus::from_db (redundant with the infallible from_disk
  given the status column's CHECK constraint)
- reuse crate::sessions::claude::transcript_cwd instead of a byte-copy
- promote string_arg/argument_error/tool_json_with_md to handlers/support
  and share them between session.rs and workflow_query.rs
- one shared shared::one_line_truncated (single ellipsis convention)
  replacing three divergent copies
- ProjectRootMatcher resolves the fixed project side's git worktree/common
  dir once, so the ingest loop stops re-resolving it per run

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* refactor: split workflow session tests

* refactor(sessions): simplify workflow indexing reuse and dedup

Reuse git_scope_exists_clauses for workflow git-scope queries, avoid
per-status string allocations, HashSet dedup in roster assembly, and
colocate one_line_truncated tests with shared helpers.

* refactor(sessions): dedupe workflow args

* refactor(sessions): extract workflow scope EXISTS predicate

* fix(workflow): satisfy branch quality gates

* fix(test): normalize CI path assertions

* fix(test): normalize Windows command paths

* fix(test): normalize Cursor hook paths

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: ScriptedAlchemy

Cargo.lock

@@ -162,6 +162,7 @@ cc = "1"
 tempfile = "3"
 sha2 = "0.11"
 hex = "0.4"
+filetime = "0.2"
 criterion = { version = "0.5", features = ["async_tokio", "html_reports"] }
 # test-util enables tokio::test(start_paused = true) so timer-driven unit
 # tests (daemon restart-grace windows) run on virtual time instead of real

Cargo.toml

@@ -151,7 +151,8 @@ per-call review, add the snippet below to `~/.cursor/permissions.json`
     "tracedecay:tracedecay_todos",
     "tracedecay:tracedecay_type_hierarchy",
     "tracedecay:tracedecay_unsafe_patterns",
-    "tracedecay:tracedecay_unused_imports"
+    "tracedecay:tracedecay_unused_imports",
+    "tracedecay:tracedecay_workflows"
   ]
 }
 ```

plugin/README-cursor.md

@@ -45,6 +45,14 @@ Climb cheapest-first; stop as soon as the question is answered.
    `branch`|`worktree`|`commit`, `value`, optional `since`/`until`, `limit`):
    find sessions active on a branch or worktree, or sessions that produced a
    commit; feed returned session ids back into grep/replay/drill-down above.
+7. **Workflow-run recovery → `tracedecay_workflows`**: recover multi-agent
+   workflow (`wf_*`) runs and their per-phase agents. List runs for a thread
+   with `session_id`, or every run on a branch/worktree/commit with
+   `branch`/`worktree`/`commit` (a run inherits its parent session's git
+   spans). Show one run's result summary + phases + agent roster with
+   `run_id`, then drill into a single agent with `run_id` + `agent_label`.
+   To read that agent's messages, scope `tracedecay_message_search` with
+   `workflow_run` (+ optional `workflow_agent`), or replay via rungs 3–4.
 
 After a compaction, if prior-session context seems missing, run this ladder
 before assuming the compacted summary is complete.

plugin/skills/managing-session-context/SKILL.md

@@ -669,6 +669,21 @@ pub enum SessionsAction {
         #[arg(long)]
         dry_run: bool,
     },
+    /// List unfinished workflow/task evidence from ingested session messages
+    Unfinished {
+        /// Maximum evidence rows
+        #[arg(long, default_value_t = 25)]
+        limit: usize,
+        /// Output as JSON
+        #[arg(long)]
+        json: bool,
+        /// Registered project id whose session store should be searched
+        #[arg(long)]
+        project_id: Option<String>,
+        /// Registered project root path or alias whose session store should be searched
+        #[arg(long, conflicts_with = "project_id")]
+        project_path: Option<String>,
+    },
 }
 
 #[derive(Subcommand)]

src/cli.rs

@@ -21,6 +21,21 @@ use crate::sessions::{
 
 const UNIX_TIMESTAMP_MILLIS_THRESHOLD: i64 = 1_000_000_000_000;
 
+/// Scopes a `tracedecay_message_search` to the agent transcripts of one
+/// workflow run, mirroring `GitScopeFilter` as a search-only concern. The run's
+/// messages are the messages of its agents (rows in `workflow_agents`); see
+/// [`GlobalDb::search_session_messages_workflow_scoped`] for the EXISTS
+/// pushdown. Serializes so the applied filter echoes cleanly into the payload.
+#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
+pub struct WorkflowScopeFilter {
+    /// The `wf_*` run whose agents' messages to keep.
+    pub run_id: String,
+    /// When set, narrows the scope to just this one agent of the run
+    /// (matched on `workflow_agents.agent_label`).
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub agent_label: Option<String>,
+}
+
 /// Total savings + call count for a project (or all projects when `project` is None).
 #[derive(Debug, Clone, serde::Serialize)]
 pub struct SavingsTotal {
@@ -1023,6 +1038,9 @@ impl GlobalDb {
         crate::sessions::git_correlation::ensure_git_correlation_schema(&db.conn)
             .await
             .ok()?;
+        crate::sessions::workflow_index::ensure_workflow_index_schema(&db.conn)
+            .await
+            .ok()?;
         // One-off self-heal: re-derive timestamps and token-usage counters
         // for legacy messages ingested before extraction existed.
         // Marker-guarded (runs once per store) and fail-open, like the LCM
@@ -3271,6 +3289,80 @@ impl GlobalDb {
         crate::sessions::git_correlation::session_ids_for_scope(&self.conn, filter).await
     }
 
+    // ── Workflow-run index ───────────────────────────────────────────
+
+    /// Inserts or updates one indexed workflow run (idempotent on `run_id`).
+    /// See [`crate::sessions::workflow_index::upsert_run`].
+    pub async fn workflow_upsert_run(
+        &self,
+        run: &crate::sessions::workflow_index::WorkflowRun,
+    ) -> Result<(), crate::sessions::workflow_index::WorkflowIndexError> {
+        crate::sessions::workflow_index::upsert_run(&self.conn, run).await
+    }
+
+    /// Inserts or updates one workflow agent (idempotent on
+    /// `(run_id, agent_label, agent_id)`).
+    /// See [`crate::sessions::workflow_index::upsert_agent`].
+    pub async fn workflow_upsert_agent(
+        &self,
+        agent: &crate::sessions::workflow_index::WorkflowAgent,
+    ) -> Result<(), crate::sessions::workflow_index::WorkflowIndexError> {
+        crate::sessions::workflow_index::upsert_agent(&self.conn, agent).await
+    }
+
+    /// Lists workflow runs spawned by one parent session, newest-first.
+    /// See [`crate::sessions::workflow_index::runs_for_session`].
+    pub async fn workflow_runs_for_session(
+        &self,
+        parent_session_id: &str,
+        limit: usize,
+    ) -> Result<
+        Vec<crate::sessions::workflow_index::WorkflowRun>,
+        crate::sessions::workflow_index::WorkflowIndexError,
+    > {
+        crate::sessions::workflow_index::runs_for_session(&self.conn, parent_session_id, limit)
+            .await
+    }
+
+    /// Fetches one workflow run by its `wf_*` id.
+    /// See [`crate::sessions::workflow_index::run_for_id`].
+    pub async fn workflow_run_for_id(
+        &self,
+        run_id: &str,
+    ) -> Result<
+        Option<crate::sessions::workflow_index::WorkflowRun>,
+        crate::sessions::workflow_index::WorkflowIndexError,
+    > {
+        crate::sessions::workflow_index::run_for_id(&self.conn, run_id).await
+    }
+
+    /// Lists the agents of one workflow run in phase order.
+    /// See [`crate::sessions::workflow_index::agents_for_run`].
+    pub async fn workflow_agents_for_run(
+        &self,
+        run_id: &str,
+        limit: usize,
+    ) -> Result<
+        Vec<crate::sessions::workflow_index::WorkflowAgent>,
+        crate::sessions::workflow_index::WorkflowIndexError,
+    > {
+        crate::sessions::workflow_index::agents_for_run(&self.conn, run_id, limit).await
+    }
+
+    /// Lists workflow runs that ran on a git branch/worktree/commit, joined
+    /// through their parent session's git spans.
+    /// See [`crate::sessions::workflow_index::runs_for_git_scope`].
+    pub async fn workflow_runs_for_git_scope(
+        &self,
+        filter: &crate::sessions::git_correlation::GitScopeFilter,
+        limit: usize,
+    ) -> Result<
+        Vec<crate::sessions::workflow_index::WorkflowRun>,
+        crate::sessions::workflow_index::WorkflowIndexError,
+    > {
+        crate::sessions::workflow_index::runs_for_git_scope(&self.conn, filter, limit).await
+    }
+
     /// Lists per-session activity windows for the historical git-correlation
     /// backfill: each row carries the session's declared `started_at`/`ended_at`
     /// plus the min/max `session_messages.timestamp`, so the caller can derive
@@ -3317,6 +3409,7 @@ impl GlobalDb {
             limit,
             filters,
             None,
+            None,
         )
         .await
     }
@@ -3342,6 +3435,37 @@ impl GlobalDb {
             limit,
             filters,
             Some(git_filter),
+            None,
+        )
+        .await
+    }
+
+    /// Like [`Self::search_session_messages_filtered`], additionally scoping
+    /// hits to the agent transcripts of one workflow run via EXISTS pushdown
+    /// against `workflow_agents`. A run's agents are matched either by the
+    /// transcript file the message came from (`workflow_agents.transcript_path
+    /// = session_messages.source_path`) or, as a fallback, by the agent's own
+    /// session id (`workflow_agents.agent_session_id = session_messages.session_id`),
+    /// so the scope holds whichever key the ingest recorded. When
+    /// `filter.agent_label` is set the scope narrows to that one agent. A call
+    /// against a store predating the workflow-index schema returns no hits.
+    pub async fn search_session_messages_workflow_scoped(
+        &self,
+        provider: Option<&str>,
+        project_key: Option<&str>,
+        query: &str,
+        limit: usize,
+        filters: SessionSearchFilters<'_>,
+        workflow_filter: &WorkflowScopeFilter,
+    ) -> Vec<SessionMessageSearchResult> {
+        self.search_session_messages_filtered_inner(
+            provider,
+            project_key,
+            query,
+            limit,
+            filters,
+            None,
+            Some(workflow_filter),
         )
         .await
     }
@@ -3354,10 +3478,19 @@ impl GlobalDb {
         limit: usize,
         filters: SessionSearchFilters<'_>,
     ) -> Vec<SessionMessageSearchResult> {
-        self.search_session_messages_filtered_inner(None, project_key, query, limit, filters, None)
-            .await
+        self.search_session_messages_filtered_inner(
+            None,
+            project_key,
+            query,
+            limit,
+            filters,
+            None,
+            None,
+        )
+        .await
     }
 
+    #[allow(clippy::too_many_arguments)] // internal fan-in of independent scope/time/git/workflow filters
     async fn search_session_messages_filtered_inner(
         &self,
         provider: Option<&str>,
@@ -3366,6 +3499,7 @@ impl GlobalDb {
         limit: usize,
         filters: SessionSearchFilters<'_>,
         git_filter: Option<&crate::sessions::git_correlation::GitScopeFilter>,
+        workflow_filter: Option<&WorkflowScopeFilter>,
     ) -> Vec<SessionMessageSearchResult> {
         // A git-scoped search against a store written before the correlation
         // schema existed can never match; report empty rather than issuing a
@@ -3379,6 +3513,16 @@ impl GlobalDb {
                 return Vec::new();
             }
         }
+        // Likewise a workflow-scoped search against a store predating the
+        // workflow-index schema can never match: short-circuit to empty rather
+        // than hitting `no such table: workflow_agents`.
+        if workflow_filter.is_some()
+            && !crate::sessions::workflow_index::tables_present(&self.conn)
+                .await
+                .unwrap_or(false)
+        {
+            return Vec::new();
+        }
         let fts_query = session_fts_query(query);
         if fts_query.is_empty() || limit == 0 {
             return Vec::new();
@@ -3455,6 +3599,24 @@ impl GlobalDb {
                 query_params.extend(predicate_values);
             }
         }
+        // Workflow-run scoping: reuse the shared EXISTS predicate (also used
+        // by future lcm/grep paths) so run/agent correlation semantics stay in
+        // one place. Renumber its `?1`, `?2`, … slots to follow the query's
+        // existing numbered placeholders, then append the bind values in order.
+        if let Some(filter) = workflow_filter {
+            let (mut predicate, predicate_values) =
+                crate::sessions::workflow_index::workflow_scope_exists_predicate(
+                    filter,
+                    "m.source_path",
+                    "m.session_id",
+                );
+            let base = query_params.len();
+            for slot in (1..=predicate_values.len()).rev() {
+                predicate = predicate.replace(&format!("?{slot}"), &format!("?{}", base + slot));
+            }
+            let _ = write!(sql, " AND {predicate}");
+            query_params.extend(predicate_values);
+        }
         for term in &literal_terms {
             query_params.push(Value::Text(term.clone()));
             let _ = write!(

src/global_db.rs

@@ -731,6 +731,9 @@ fn should_skip_startup_maintenance(command: &Commands) -> bool {
             | Commands::Lsp { .. }
             | Commands::Doctor { .. }
             | Commands::Analytics { .. }
+            | Commands::Sessions {
+                action: SessionsAction::Unfinished { .. },
+            }
             | Commands::Migrate { .. }
             | Commands::Projects { .. }
             | Commands::HookPreToolUse
@@ -806,6 +809,9 @@ fn should_skip_agent_install_maintenance(command: &Commands) -> bool {
             | Commands::Migrate { .. }
             | Commands::Projects { .. }
             | Commands::Tool { .. }
+            | Commands::Sessions {
+                action: SessionsAction::Unfinished { .. },
+            }
             | Commands::Daemon { .. }
     )
 }

src/main.rs

@@ -309,6 +309,7 @@ pub fn get_tool_definitions() -> Vec<ToolDefinition> {
         def_dashboard(),
         def_message_search(),
         def_sessions_for(),
+        def_workflows(),
         def_lcm_status(),
         def_lcm_doctor(),
         def_lcm_load_session(),
@@ -2463,7 +2464,9 @@ fn def_message_search() -> ToolDefinition {
                 },
                 "branch": git_scope_branch_schema(),
                 "worktree": git_scope_worktree_schema(),
-                "commit": git_scope_commit_schema()
+                "commit": git_scope_commit_schema(),
+                "workflow_run": workflow_run_scope_schema(),
+                "workflow_agent": workflow_agent_scope_schema()
             },
             "required": ["query"]
         }),
@@ -2534,6 +2537,65 @@ fn def_sessions_for() -> ToolDefinition {
     )
 }
 
+fn def_workflows() -> ToolDefinition {
+    def(
+        "tracedecay_workflows",
+        "Workflow Runs",
+        "Recover Claude Code workflow runs (multi-agent `wf_*` orchestrations) and their per-phase agents from the active project. Three modes, chosen by which argument is set: (1) list runs for a parent thread via session_id, or every run on a branch/worktree/commit via branch/worktree/commit (a run inherits its parent session's git spans); (2) show one run's result summary, phases, and agent roster via run_id; (3) drill into one agent's transcript via run_id + agent_label. Read-only; runs that never ran leave no rows.",
+        json!({
+            "type": "object",
+            "properties": {
+                "session_id": {
+                    "type": "string",
+                    "description": "Parent thread/session id: list the workflow runs it spawned (newest first). Mutually exclusive with run_id and the git filters."
+                },
+                "run_id": {
+                    "type": "string",
+                    "description": "A `wf_*` run id: show that run's summary, phases, and agents. Combine with agent_label to drill into one agent."
+                },
+                "agent_label": {
+                    "type": "string",
+                    "description": "With run_id, drill into a single agent of that run by its label (e.g. 'mine:claude-transcripts')."
+                },
+                "branch": {
+                    "type": "string",
+                    "description": "List workflow runs whose parent session was active on this git branch (via the session-git correlation index)."
+                },
+                "worktree": {
+                    "type": "string",
+                    "description": "List workflow runs whose parent session was active in this git worktree root path."
+                },
+                "commit": {
+                    "type": "string",
+                    "description": "List workflow runs whose parent session was attributed to this commit sha (full or >=6-char hex prefix)."
+                },
+                "limit": {
+                    "type": "integer",
+                    "minimum": 1,
+                    "maximum": 100,
+                    "description": "Maximum runs or agents to return (default: 20)."
+                }
+            }
+        }),
+    )
+}
+
+/// Optional `workflow_run` narrowing filter shared by `tracedecay_message_search`:
+/// scopes hits to the transcripts of one workflow run's agents.
+fn workflow_run_scope_schema() -> Value {
+    json!({
+        "type": "string",
+        "description": "Optional workflow run id (`wf_*`) filter: only messages from sessions that spawned this workflow run (via the workflow-run index). Pair with agent_label to scope to one agent."
+    })
+}
+
+fn workflow_agent_scope_schema() -> Value {
+    json!({
+        "type": "string",
+        "description": "Optional workflow agent label filter, used with workflow_run to scope to a single agent of that run."
+    })
+}
+
 fn lcm_storage_scope_schema() -> Value {
     json!({
         "type": "string",