feat(subject): v0.6.29 — BaaS dynamic-kind enqueue/run (--subject-id) (#296)

Shooksie · web-flow · commit efd26890ffbb · 2026-06-30T07:58:02.000-06:00
* feat: BaaS dynamic-kind dispatch on queue enqueue + workflow run (--subject-id)

Add a generic --subject-id flag to `queue enqueue` and `workflow run` so a
subject of an arbitrary kind (e.g. kind=blog, id BLOG-001) can be queued/run.
Previously only --task-id/--requirement-id/--title existed, so a non-task
subject was coerced to kind=task and failed subject resolution
("id 'BLOG-001' is a 'blog' subject, not 'task'").

- New shared resolver (services/operations/subject_id_dispatch.rs): a
  qualified id (blog:BLOG-001) trusts the explicit kind; a bare id is resolved
  by probing the installed subject backends (router) for the owning kind. The
  resolved kind is preserved on the dispatch so the subject is leased/run via
  &lt;kind&gt;/get rather than coerced to task. task/requirement keep their strict
  flags unchanged.
- queue enqueue + workflow run (sync and detached) consume the kind-correct
  SubjectDispatch. The detached reattach builder now rebuilds a subject_dispatch
  for generic kinds (the runner protocol requires it for non-task/requirement
  subjects), and the daemon queue-lease -&gt; runner command emits --subject-id
  (kind-qualified) for generic kinds instead of collapsing them to --title.
- MCP: animus.queue.enqueue and animus.workflow.run expose a subject_id param
  mirroring the CLI flag, with agent-facing guidance to use it for BaaS dynamic
  kinds.
- Docs: cli/index.md, mcp-tools.md, guides/agents.md.

NOTE: the queue-lease -&gt; runner CLI leg emits --subject-id only for generic
kinds; the workflow_runner plugin must accept --subject-id for that path
(task/requirement/custom dispatch is unchanged for existing runners). The
detached `workflow run --subject-id` + MCP path works end-to-end via the
plugin workflow/execute RPC.

* chore(release): v0.6.29 — BaaS dynamic-kind dispatch (--subject-id + kind resolution on enqueue/run + MCP subject_id)
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/crates/orchestrator-cli/Cargo.toml b/crates/orchestrator-cli/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "orchestrator-cli"
-version = "0.6.28"
+version = "0.6.29"
 edition = "2021"
 license = "Elastic-2.0"
 default-run = "animus"
diff --git a/crates/orchestrator-cli/src/cli_types/mod.rs b/crates/orchestrator-cli/src/cli_types/mod.rs
@@ -244,6 +244,59 @@ mod tests {
         }
     }
 
+    #[test]
+    fn queue_enqueue_accepts_subject_id() {
+        let cli = Cli::try_parse_from(["animus", "queue", "enqueue", "--subject-id", "blog:BLOG-001"])
+            .expect("queue enqueue --subject-id should parse");
+        match cli.command {
+            Command::Queue { command: QueueCommand::Enqueue(args) } => {
+                assert_eq!(args.subject_id.as_deref(), Some("blog:BLOG-001"));
+                assert!(args.task_id.is_none());
+            }
+            other => panic!("expected queue enqueue, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn queue_enqueue_subject_id_conflicts_with_kind_flags() {
+        for other in [["--task-id", "TASK-1"], ["--requirement-id", "REQ-1"], ["--title", "x"]] {
+            let error = Cli::try_parse_from([
+                "animus",
+                "queue",
+                "enqueue",
+                "--subject-id",
+                "blog:BLOG-001",
+                other[0],
+                other[1],
+            ])
+            .expect_err("--subject-id must be mutually exclusive with the kind flags");
+            assert_eq!(error.kind(), ErrorKind::ArgumentConflict, "flag {}: {error}", other[0]);
+        }
+    }
+
+    #[test]
+    fn workflow_run_accepts_subject_id() {
+        let cli = Cli::try_parse_from(["animus", "workflow", "run", "--subject-id", "blog:BLOG-001"])
+            .expect("workflow run --subject-id should parse");
+        match cli.command {
+            Command::Workflow { command: WorkflowCommand::Run(args) } => {
+                assert_eq!(args.subject_id.as_deref(), Some("blog:BLOG-001"));
+                assert!(args.task_id.is_none());
+            }
+            other => panic!("expected workflow run, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn workflow_run_subject_id_conflicts_with_kind_flags() {
+        for other in [["--task-id", "TASK-1"], ["--requirement-id", "REQ-1"], ["--title", "x"]] {
+            let error =
+                Cli::try_parse_from(["animus", "workflow", "run", "--subject-id", "blog:BLOG-001", other[0], other[1]])
+                    .expect_err("--subject-id must be mutually exclusive with the kind flags");
+            assert_eq!(error.kind(), ErrorKind::ArgumentConflict, "flag {}: {error}", other[0]);
+        }
+    }
+
     #[test]
     fn daemon_metrics_bare_defaults_to_display() {
         let cli = Cli::try_parse_from(["animus", "daemon", "metrics"]).expect("bare daemon metrics should parse");
diff --git a/crates/orchestrator-cli/src/cli_types/queue_types.rs b/crates/orchestrator-cli/src/cli_types/queue_types.rs
@@ -48,6 +48,13 @@ pub(crate) struct QueueEnqueueArgs {
         help = "Custom subject title for ad-hoc dispatches. Mutually exclusive with --task-id / --requirement-id."
     )]
     pub(crate) title: Option<String>,
+    #[arg(
+        long = "subject-id",
+        value_name = "SUBJECT_ID",
+        group = "subject",
+        help = "Generic subject to enqueue for any kind (BaaS dynamic kinds like blog/post). Accepts a qualified id (blog:BLOG-001 — kind trusted; the recommended form) or a bare id (BLOG-001 — kind probed across backends that declare concrete kinds; pure catch-all dynamic backends require the qualified form). Mutually exclusive with --task-id / --requirement-id / --title."
+    )]
+    pub(crate) subject_id: Option<String>,
     #[arg(long, value_name = "TEXT", help = "Custom subject description (used with --title).")]
     pub(crate) description: Option<String>,
     #[arg(long = "workflow-ref", value_name = "WORKFLOW_REF", help = "Optional YAML workflow reference override.")]
diff --git a/crates/orchestrator-cli/src/cli_types/workflow_types.rs b/crates/orchestrator-cli/src/cli_types/workflow_types.rs
@@ -287,6 +287,13 @@ pub(crate) struct WorkflowRunArgs {
     pub(crate) requirement_id: Option<String>,
     #[arg(long, value_name = "TITLE", group = "subject", help = "Custom workflow title for freeform execution.")]
     pub(crate) title: Option<String>,
+    #[arg(
+        long = "subject-id",
+        value_name = "SUBJECT_ID",
+        group = "subject",
+        help = "Generic subject to run the workflow for, any kind (BaaS dynamic kinds like blog/post). Accepts a qualified id (blog:BLOG-001 — kind trusted; the recommended form) or a bare id (BLOG-001 — kind probed across backends that declare concrete kinds; pure catch-all dynamic backends require the qualified form). Mutually exclusive with --task-id / --requirement-id / --title."
+    )]
+    pub(crate) subject_id: Option<String>,
     #[arg(long, value_name = "TEXT", help = "Custom workflow description (used with --title).")]
     pub(crate) description: Option<String>,
     #[arg(
diff --git a/crates/orchestrator-cli/src/services/operations.rs b/crates/orchestrator-cli/src/services/operations.rs
@@ -24,6 +24,7 @@ mod ops_trigger;
 mod ops_update;
 mod ops_web;
 mod ops_workflow;
+mod subject_id_dispatch;
 
 pub(crate) use ops_auth::*;
 use ops_common::*;
diff --git a/crates/orchestrator-cli/src/services/operations/ops_mcp/queue_command_args.rs b/crates/orchestrator-cli/src/services/operations/ops_mcp/queue_command_args.rs
@@ -5,6 +5,7 @@ pub(super) fn build_queue_enqueue_args(input: &QueueEnqueueInput) -> Vec<String>
     push_opt(&mut args, "--task-id", input.task_id.clone());
     push_opt(&mut args, "--requirement-id", input.requirement_id.clone());
     push_opt(&mut args, "--title", input.title.clone());
+    push_opt(&mut args, "--subject-id", input.subject_id.clone());
     push_opt(&mut args, "--description", input.description.clone());
     push_opt(&mut args, "--workflow-ref", input.workflow_ref.clone());
     push_opt(&mut args, "--input-json", input.input_json.clone());
diff --git a/crates/orchestrator-cli/src/services/operations/ops_mcp/queue_inputs.rs b/crates/orchestrator-cli/src/services/operations/ops_mcp/queue_inputs.rs
@@ -8,6 +8,13 @@ pub(super) struct QueueEnqueueInput {
     pub(super) requirement_id: Option<String>,
     #[serde(default)]
     pub(super) title: Option<String>,
+    /// For subjects that are NOT kind=task/requirement (BaaS dynamic kinds like
+    /// blog/post/etc.), pass `subject_id` (the kernel resolves the kind)
+    /// instead of `task_id`. Accepts a qualified id (`blog:BLOG-001` — kind
+    /// trusted) or a bare id (`BLOG-001` — kind resolved via the subject
+    /// router). Mutually exclusive with `task_id` / `requirement_id` / `title`.
+    #[serde(default)]
+    pub(super) subject_id: Option<String>,
     #[serde(default)]
     pub(super) description: Option<String>,
     #[serde(default)]
diff --git a/crates/orchestrator-cli/src/services/operations/ops_mcp/queue_tools.rs b/crates/orchestrator-cli/src/services/operations/ops_mcp/queue_tools.rs
@@ -22,7 +22,7 @@ impl AoMcpServer {
 
     #[tool(
         name = "animus.queue.enqueue",
-        description = "Enqueue a subject dispatch. Purpose: Add a SubjectDispatch to the daemon queue using a task, requirement, or custom subject plus optional workflow/input override. Prerequisites: Task subjects must exist; custom subjects require a title. Example: {\"task_id\": \"TASK-001\", \"workflow_ref\": \"ops\"}. Sequencing: Use animus.queue.list to inspect position or animus.queue.reorder to adjust ordering.",
+        description = "Enqueue a subject dispatch. Purpose: Add a SubjectDispatch to the daemon queue using a task, requirement, custom, or arbitrary-kind subject plus optional workflow/input override. Prerequisites: Subjects must exist; custom subjects require a title. For subjects that are NOT kind=task/requirement (BaaS dynamic kinds like blog/post/etc.), pass subject_id (the kernel resolves the kind) instead of task_id. Example: {\"task_id\": \"TASK-001\", \"workflow_ref\": \"ops\"} or {\"subject_id\": \"blog:BLOG-001\"}. Sequencing: Use animus.queue.list to inspect position or animus.queue.reorder to adjust ordering.",
         input_schema = ao_schema_for_type::<QueueEnqueueInput>()
     )]
     async fn ao_queue_enqueue(&self, params: Parameters<QueueEnqueueInput>) -> Result<CallToolResult, McpError> {
diff --git a/crates/orchestrator-cli/src/services/operations/ops_mcp/tests.rs b/crates/orchestrator-cli/src/services/operations/ops_mcp/tests.rs
@@ -865,6 +865,7 @@ fn build_queue_enqueue_args_includes_optional_fields() {
         task_id: Some("TASK-123".to_string()),
         requirement_id: None,
         title: None,
+        subject_id: None,
         description: None,
         workflow_ref: Some("ops".to_string()),
         input_json: Some("{\"mode\":\"fast\"}".to_string()),
diff --git a/crates/orchestrator-cli/src/services/operations/ops_mcp/workflow_inputs.rs b/crates/orchestrator-cli/src/services/operations/ops_mcp/workflow_inputs.rs
@@ -32,6 +32,13 @@ pub(super) struct WorkflowRunInput {
     pub(super) requirement_id: Option<String>,
     #[serde(default)]
     pub(super) title: Option<String>,
+    /// For subjects that are NOT kind=task/requirement (BaaS dynamic kinds like
+    /// blog/post/etc.), pass `subject_id` (the kernel resolves the kind)
+    /// instead of `task_id`. Accepts a qualified id (`blog:BLOG-001` — kind
+    /// trusted) or a bare id (`BLOG-001` — kind resolved via the subject
+    /// router). Mutually exclusive with `task_id` / `requirement_id` / `title`.
+    #[serde(default)]
+    pub(super) subject_id: Option<String>,
     #[serde(default)]
     pub(super) description: Option<String>,
     #[serde(default)]
diff --git a/crates/orchestrator-cli/src/services/operations/ops_mcp/workflow_runtime_tools.rs b/crates/orchestrator-cli/src/services/operations/ops_mcp/workflow_runtime_tools.rs
@@ -20,7 +20,7 @@ impl AoMcpServer {
 
     #[tool(
         name = "animus.workflow.run",
-        description = "Run a workflow for a task. Purpose: Execute a workflow to complete task phases automatically. Prerequisites: Task should exist (use animus.subject.get with {\"kind\":\"task\"} to verify). Example: {\"task_id\": \"TASK-001\"} or {\"task_id\": \"TASK-001\", \"workflow_ref\": \"default\"}. Sequencing: Use animus.subject.status with {\"kind\":\"task\", ...} to track progress, animus.workflow.get to monitor, and animus.workflow.pause/resume/cancel for control.",
+        description = "Run a workflow for a subject. Purpose: Execute a workflow to complete subject phases automatically. Prerequisites: Subject should exist. For subjects that are NOT kind=task/requirement (BaaS dynamic kinds like blog/post/etc.), pass subject_id (the kernel resolves the kind) instead of task_id. Example: {\"task_id\": \"TASK-001\"} or {\"subject_id\": \"blog:BLOG-001\"} or {\"subject_id\": \"blog:BLOG-001\", \"workflow_ref\": \"draft-post\"}. Sequencing: Use animus.subject.status to track progress, animus.workflow.get to monitor, and animus.workflow.pause/resume/cancel for control.",
         input_schema = ao_schema_for_type::<WorkflowRunInput>()
     )]
     async fn ao_workflow_run(&self, params: Parameters<WorkflowRunInput>) -> Result<CallToolResult, McpError> {
@@ -30,6 +30,7 @@ impl AoMcpServer {
         push_opt(&mut args, "--task-id", input.task_id);
         push_opt(&mut args, "--requirement-id", input.requirement_id);
         push_opt(&mut args, "--title", input.title);
+        push_opt(&mut args, "--subject-id", input.subject_id);
         push_opt(&mut args, "--description", input.description);
         push_opt(&mut args, "--input-json", input.input_json);
         self.run_tool("animus.workflow.run", args, input.project_root).await
diff --git a/crates/orchestrator-cli/src/services/operations/ops_queue.rs b/crates/orchestrator-cli/src/services/operations/ops_queue.rs
@@ -61,6 +61,32 @@ async fn resolve_enqueue_dispatch(
     }
 }
 
+/// Resolve a generic `--subject-id` enqueue into a kind-correct
+/// [`SubjectDispatch`]. The subject's real kind is taken from the qualified
+/// prefix or discovered by probing the installed subject backends, so a
+/// `kind=blog` subject is stored as a `blog` dispatch (and leased/resolved via
+/// `blog/get`) rather than coerced to `task`.
+async fn resolve_enqueue_dispatch_for_subject_id(
+    project_root: &str,
+    subject_id: &str,
+    workflow_ref: Option<String>,
+    input: Option<serde_json::Value>,
+) -> Result<SubjectDispatch> {
+    use super::subject_id_dispatch::{resolve_subject_id_ref, RouterSubjectProbe};
+    let probe = RouterSubjectProbe::discover(std::path::Path::new(project_root)).await?;
+    let subject_ref = resolve_subject_id_ref(subject_id, &probe).await?;
+    let workflow_ref = workflow_ref.unwrap_or_else(|| {
+        load_workflow_config_or_default(std::path::Path::new(project_root)).config.default_workflow_ref
+    });
+    Ok(SubjectDispatch::for_subject_with_metadata(
+        subject_ref,
+        workflow_ref,
+        "manual-queue-enqueue",
+        chrono::Utc::now(),
+    )
+    .with_input(input))
+}
+
 fn queue_plugin_required(operation: &str) -> anyhow::Error {
     // Same human-readable text as the previous bare `anyhow!` constructor
     // (which classified as Internal via the message fallback) — this only
@@ -109,17 +135,25 @@ pub(crate) async fn handle_queue(
         }
         QueueCommand::Enqueue(args) => {
             let input = args.input_json.clone().map(|value| serde_json::from_str(&value)).transpose()?;
-            let dispatch = resolve_enqueue_dispatch(
-                hub.clone(),
-                project_root,
-                args.task_id.clone(),
-                args.requirement_id.clone(),
-                args.title.clone(),
-                args.description.clone(),
-                args.workflow_ref.clone(),
-                input,
-            )
-            .await?;
+            let dispatch = if let Some(subject_id) = args.subject_id.clone() {
+                // Generic BaaS path: resolve the subject's real kind (qualified
+                // prefix or router probe) so a non-task/requirement subject
+                // dispatches under its own kind instead of being coerced to task.
+                resolve_enqueue_dispatch_for_subject_id(project_root, &subject_id, args.workflow_ref.clone(), input)
+                    .await?
+            } else {
+                resolve_enqueue_dispatch(
+                    hub.clone(),
+                    project_root,
+                    args.task_id.clone(),
+                    args.requirement_id.clone(),
+                    args.title.clone(),
+                    args.description.clone(),
+                    args.workflow_ref.clone(),
+                    input,
+                )
+                .await?
+            };
 
             let run_at = args.run_at.as_deref().map(resolve_run_at).transpose()?;
 
diff --git a/crates/orchestrator-cli/src/services/operations/ops_workflow/execute.rs b/crates/orchestrator-cli/src/services/operations/ops_workflow/execute.rs
@@ -15,6 +15,13 @@ pub(crate) struct WorkflowExecuteArgs {
     pub(crate) task_id: Option<String>,
     pub(crate) requirement_id: Option<String>,
     pub(crate) title: Option<String>,
+    /// Generic, kind-correct subject dispatch for the BaaS `--subject-id` path.
+    /// When set (and not re-attaching to an existing workflow), it is relayed
+    /// verbatim on the runner request as `subject_dispatch` — the protocol
+    /// requires generic (non-task/requirement) subject backends to use the
+    /// dispatch envelope — so the subject resolves under its real kind via
+    /// `<kind>/get` instead of being coerced to `task`.
+    pub(crate) subject_dispatch: Option<protocol::SubjectDispatch>,
     pub(crate) description: Option<String>,
     pub(crate) workflow_ref: Option<String>,
     pub(crate) phase: Option<String>,
@@ -115,7 +122,7 @@ pub(crate) async fn handle_workflow_execute(
     } else {
         workflow_proto::WorkflowExecuteRequest {
             workflow_id: None,
-            subject_dispatch: None,
+            subject_dispatch: args.subject_dispatch.clone(),
             subject_ref: None,
             task_id: args.task_id.clone(),
             requirement_id: args.requirement_id.clone(),
diff --git a/crates/orchestrator-cli/src/services/operations/ops_workflow/mod.rs b/crates/orchestrator-cli/src/services/operations/ops_workflow/mod.rs
@@ -52,6 +52,19 @@ fn print_workflow_list_table(items: &[protocol::orchestrator::OrchestratorWorkfl
     render_table(&["ID", "WORKFLOW", "STATUS", "TASK", "STARTED"], &rows);
 }
 
+/// Resolve a generic `--subject-id` value to a kind-correct [`SubjectRef`] by
+/// probing the installed subject backends (the `*` catch-all serves BaaS
+/// dynamic kinds). Qualified ids (`blog:BLOG-001`) trust the explicit kind;
+/// bare ids are resolved to their owning kind.
+async fn resolve_subject_id_ref_via_router(
+    project_root: &str,
+    subject_id: &str,
+) -> Result<protocol::orchestrator::SubjectRef> {
+    use super::subject_id_dispatch::{resolve_subject_id_ref, RouterSubjectProbe};
+    let probe = RouterSubjectProbe::discover(std::path::Path::new(project_root)).await?;
+    resolve_subject_id_ref(subject_id, &probe).await
+}
+
 #[allow(clippy::too_many_arguments)]
 async fn resolve_workflow_run_dispatch(
     hub: Arc<dyn ServiceHub>,
@@ -443,6 +456,34 @@ pub(crate) async fn handle_workflow(
             // stripped so a foreign-kind qualifier is not silently rewritten.
             args.task_id = args.task_id.map(|id| crate::bare_task_id(&id));
             let workflow_ref = args.pipeline.clone();
+            // Generic BaaS subject path: resolve the subject's real kind once
+            // (qualified prefix or router probe) and build a kind-correct
+            // dispatch carrying the resolved SubjectRef (kind=blog,
+            // id=blog:BLOG-001, ...) plus this run's workflow_ref / input / vars.
+            // The runner protocol requires generic (non-task/requirement)
+            // subjects to travel as a `subject_dispatch`, so both the sync and
+            // detached paths consume this single envelope rather than coercing
+            // the subject to task.
+            let subject_id_dispatch = match args.subject_id.clone() {
+                Some(sid) => {
+                    let subject_ref = resolve_subject_id_ref_via_router(project_root, &sid).await?;
+                    let resolved_ref =
+                        workflow_ref.clone().unwrap_or_else(|| default_project_workflow_ref(project_root));
+                    let input = args.input_json.as_deref().map(serde_json::from_str).transpose()?;
+                    let vars = parse_workflow_vars(&args.vars)?;
+                    Some(
+                        protocol::SubjectDispatch::for_subject_with_metadata(
+                            subject_ref,
+                            resolved_ref,
+                            "manual-cli-run",
+                            Utc::now(),
+                        )
+                        .with_input(input)
+                        .with_vars(vars),
+                    )
+                }
+                None => None,
+            };
             // Transport-asserted caller identity for this run. The transport
             // (e.g. the portal) authenticates the user and passes
             // `--actor-json`; the kernel relays it verbatim to the runner /
@@ -455,6 +496,7 @@ pub(crate) async fn handle_workflow(
                     task_id: args.task_id,
                     requirement_id: args.requirement_id,
                     title: args.title,
+                    subject_dispatch: subject_id_dispatch.clone(),
                     description: args.description,
                     workflow_ref: workflow_ref.clone(),
                     phase: args.phase,
@@ -521,20 +563,28 @@ pub(crate) async fn handle_workflow(
                         }
                     }
                 }
-                let dispatch = match args.input_json {
-                    Some(raw) => resolve_workflow_run_dispatch_from_raw_input(hub.clone(), project_root, &raw).await?,
-                    None => {
-                        resolve_workflow_run_dispatch(
-                            hub.clone(),
-                            project_root,
-                            args.task_id,
-                            args.requirement_id,
-                            args.title,
-                            args.description,
-                            workflow_ref,
-                            parsed_vars,
-                        )
-                        .await?
+                let dispatch = if let Some(subject_dispatch) = subject_id_dispatch {
+                    // Generic subject: the kind-correct dispatch was built up
+                    // front; the detached runner resolves it via `<kind>/get`.
+                    subject_dispatch
+                } else {
+                    match args.input_json {
+                        Some(raw) => {
+                            resolve_workflow_run_dispatch_from_raw_input(hub.clone(), project_root, &raw).await?
+                        }
+                        None => {
+                            resolve_workflow_run_dispatch(
+                                hub.clone(),
+                                project_root,
+                                args.task_id,
+                                args.requirement_id,
+                                args.title,
+                                args.description,
+                                workflow_ref,
+                                parsed_vars,
+                            )
+                            .await?
+                        }
                     }
                 };
                 // Post-v0.5 there is no in-process executor: a bare
diff --git a/crates/orchestrator-cli/src/services/operations/ops_workflow/phases.rs b/crates/orchestrator-cli/src/services/operations/ops_workflow/phases.rs
diff --git a/crates/orchestrator-cli/src/services/operations/subject_id_dispatch.rs b/crates/orchestrator-cli/src/services/operations/subject_id_dispatch.rs
diff --git a/crates/orchestrator-daemon-runtime/src/dispatch/build_runner_command_from_dispatch.rs b/crates/orchestrator-daemon-runtime/src/dispatch/build_runner_command_from_dispatch.rs
diff --git a/docs/guides/agents.md b/docs/guides/agents.md
diff --git a/docs/reference/cli/index.md b/docs/reference/cli/index.md
diff --git a/docs/reference/mcp-tools.md b/docs/reference/mcp-tools.md
