feat(actor): add plan_subplan tool and decision emission

Add builtin/plan-subplan tool for strategy actors to emit SUBPLAN_SPAWN
or SUBPLAN_PARALLEL_SPAWN decisions when decomposing a plan into child
plans. Implements all acceptance criteria from issue #198:

- SubplanPayload (Pydantic) validates goal, resource_scopes/project_ref
  (at least one required), merge_strategy, max_parallel (1-50), parallel
  flag, dependencies, and context_view override.
- Defaults: merge_strategy=git_three_way, max_parallel=5, parallel=False,
  dependencies=[]. Omitted fields inherit sensible values automatically.
- make_plan_subplan_spec(decision_service=None) factory supports optional
  DecisionService injection for persistent decision recording.
- _build_rationale() generates human-readable rationale text (goal, scope,
  execution mode, dependencies, context_view) surfaced in plan explain.
- register_subplan_tool() added to tool/builtins/__init__.py for bulk
  registration. PLAN_SUBPLAN_SPEC exported as the default ready-to-use spec.
- Actor YAML example (examples/actors/strategy_with_subplan.yaml) with
  annotated serial and parallel spawn payload examples.
- Behave BDD: 20 scenarios, 70 steps covering validation, defaults,
  decision type, rationale, service injection, registry, and ToolRunner.
- Robot Framework: 9 smoke tests via robot/plan_subplan_tool.robot.
- ASV benchmarks: benchmarks/subplan_actor_tool_bench.py (5 suites).
- Coverage: 100% on subplan_tool.py. Lint, typecheck, and security clean.

ISSUES CLOSED: #198

CHANGELOG.md

@@ -2,14 +2,18 @@
 
 ## Unreleased
 
-### feat(actor): extend hierarchical actor YAML schema and loader
-
+- Added `builtin/plan-subplan` tool for strategy actors to emit `SUBPLAN_SPAWN` or
+  `SUBPLAN_PARALLEL_SPAWN` decisions. Validates payload via `SubplanPayload` (Pydantic),
+  applies defaults (merge strategy, max_parallel, dependencies), generates rationale text,
+  and optionally persists the `Decision` via an injected `DecisionService`. Includes
+  `register_subplan_tool`, `make_plan_subplan_spec` factory, Behave unit tests, Robot
+  Framework integration smoke tests, ASV benchmarks, and an actor YAML example. (#198)
 - Extended actor YAML schema with hierarchical graph support: per-node LSP bindings (`lsp_binding`), tool-source references (`tool_sources`), and subgraph `actor_ref`.
 - Added graph reachability validation — all nodes must be reachable from `entry_node` via edges or conditional routing targets.
 - Improved loader error reporting with YAML line/column positions and Pydantic field-path hints.
 - Added `docs/reference/actor_config.md` — practical configuration reference with hierarchical examples and error cases.
 - Fixed `examples/actors/graph_workflow.yaml` to use `actor_ref` instead of deprecated `actor_path`.
-- Added Robot smoke test for loading hierarchical actor YAML via `ActorLoader.discover()`.
+- Added Robot smoke test for loading hierarchical actor YAML via `ActorLoader.discover()` (#157).
 
 - Added decision persistence layer with DecisionRepository, DecisionModel, Alembic migration,
   tree queries (BFS traversal, path-to-root), superseded lookup, and ordered decision path

benchmarks/subplan_actor_tool_bench.py

@@ -0,0 +1,206 @@
+"""ASV benchmarks for the builtin/plan-subplan actor tool.
+
+Measures:
+- SubplanPayload construction and validation
+- Tool handler invocation (serial and parallel spawn)
+- Decision emission with injected stub service
+- Tool registration overhead
+- ToolRunner round-trip execution
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from typing import Any
+
+try:
+    from cleveragents.domain.models.core.decision import Decision
+    from cleveragents.tool.builtins import register_subplan_tool
+    from cleveragents.tool.builtins.subplan_tool import (
+        PLAN_SUBPLAN_SPEC,
+        SubplanPayload,
+        make_plan_subplan_spec,
+    )
+    from cleveragents.tool.registry import ToolRegistry
+    from cleveragents.tool.runner import ToolRunner
+except ModuleNotFoundError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
+    from cleveragents.domain.models.core.decision import Decision
+    from cleveragents.tool.builtins import register_subplan_tool
+    from cleveragents.tool.builtins.subplan_tool import (
+        PLAN_SUBPLAN_SPEC,
+        SubplanPayload,
+        make_plan_subplan_spec,
+    )
+    from cleveragents.tool.registry import ToolRegistry
+    from cleveragents.tool.runner import ToolRunner
+
+_PLAN_ID = "01ARZ3NDEKTSV4RRFFQ69G5FAV"
+
+
+def _serial_inputs(**extra: Any) -> dict[str, Any]:
+    return {
+        "goal": "Refactor authentication module",
+        "plan_id": _PLAN_ID,
+        "sequence_number": 0,
+        "resource_scopes": ["src/auth/"],
+        "parallel": False,
+        **extra,
+    }
+
+
+def _parallel_inputs(**extra: Any) -> dict[str, Any]:
+    return {
+        "goal": "Run full test suite in parallel",
+        "plan_id": _PLAN_ID,
+        "sequence_number": 1,
+        "resource_scopes": ["tests/"],
+        "parallel": True,
+        "max_parallel": 4,
+        "dependencies": [],
+        **extra,
+    }
+
+
+# ---------------------------------------------------------------------------
+# Stub service (no DB)
+# ---------------------------------------------------------------------------
+
+
+class _StubService:
+    def __init__(self) -> None:
+        self.recorded: list[Decision] = []
+
+    def record_decision(self, decision: Decision) -> Decision:
+        self.recorded.append(decision)
+        return decision
+
+
+# ---------------------------------------------------------------------------
+# SubplanPayload validation benchmarks
+# ---------------------------------------------------------------------------
+
+
+class SubplanPayloadSuite:
+    """Benchmark SubplanPayload construction and model validation."""
+
+    def time_serial_payload_construction(self) -> None:
+        SubplanPayload(
+            goal="Refactor auth",
+            plan_id=_PLAN_ID,
+            sequence_number=0,
+            resource_scopes=["src/auth/"],
+        )
+
+    def time_parallel_payload_construction(self) -> None:
+        SubplanPayload(
+            goal="Run tests",
+            plan_id=_PLAN_ID,
+            sequence_number=1,
+            resource_scopes=["tests/"],
+            parallel=True,
+            max_parallel=8,
+            dependencies=["01ARZ3NDEKTSV4RRFFQ69G5FAW"],
+        )
+
+    def time_payload_with_all_fields(self) -> None:
+        SubplanPayload(
+            goal="Full config",
+            plan_id=_PLAN_ID,
+            sequence_number=2,
+            resource_scopes=["src/", "tests/"],
+            project_ref="projects/main",
+            parallel=True,
+            max_parallel=10,
+            merge_strategy="fail_on_conflict",  # type: ignore[arg-type]
+            dependencies=["01ARZ3NDEKTSV4RRFFQ69G5FAW"],
+            context_view="executor",
+        )
+
+
+# ---------------------------------------------------------------------------
+# Handler invocation benchmarks
+# ---------------------------------------------------------------------------
+
+
+class SubplanHandlerSuite:
+    """Benchmark the tool handler for serial and parallel spawns."""
+
+    def setup(self) -> None:
+        self.handler = PLAN_SUBPLAN_SPEC.handler
+
+    def time_serial_spawn(self) -> None:
+        self.handler(_serial_inputs())
+
+    def time_parallel_spawn(self) -> None:
+        self.handler(_parallel_inputs())
+
+    def time_serial_with_project_ref(self) -> None:
+        self.handler(
+            {
+                "goal": "Migrate database schema",
+                "plan_id": _PLAN_ID,
+                "sequence_number": 3,
+                "project_ref": "projects/db",
+                "parallel": False,
+            }
+        )
+
+    def time_serial_with_context_view(self) -> None:
+        self.handler(_serial_inputs(context_view="executor"))
+
+
+# ---------------------------------------------------------------------------
+# Decision service injection benchmark
+# ---------------------------------------------------------------------------
+
+
+class SubplanServiceSuite:
+    """Benchmark handler with injected decision service."""
+
+    def setup(self) -> None:
+        self._stub = _StubService()
+        spec = make_plan_subplan_spec(decision_service=self._stub)
+        self.handler = spec.handler
+
+    def teardown(self) -> None:
+        self._stub.recorded.clear()
+
+    def time_handler_with_service(self) -> None:
+        self.handler(_serial_inputs())
+
+
+# ---------------------------------------------------------------------------
+# Registration and ToolRunner benchmarks
+# ---------------------------------------------------------------------------
+
+
+class SubplanRegistrationSuite:
+    """Benchmark tool registration overhead."""
+
+    def time_register_subplan_tool(self) -> None:
+        registry = ToolRegistry()
+        register_subplan_tool(registry)
+
+    def time_make_plan_subplan_spec(self) -> None:
+        make_plan_subplan_spec()
+
+    def time_make_plan_subplan_spec_with_service(self) -> None:
+        stub = _StubService()
+        make_plan_subplan_spec(decision_service=stub)
+
+
+class SubplanRunnerSuite:
+    """Benchmark ToolRunner round-trip for builtin/plan-subplan."""
+
+    def setup(self) -> None:
+        registry = ToolRegistry()
+        register_subplan_tool(registry)
+        self.runner = ToolRunner(registry)
+
+    def time_runner_serial_spawn(self) -> None:
+        self.runner.execute("builtin/plan-subplan", _serial_inputs())
+
+    def time_runner_parallel_spawn(self) -> None:
+        self.runner.execute("builtin/plan-subplan", _parallel_inputs())

examples/actors/strategy_with_subplan.yaml

@@ -0,0 +1,66 @@
+# Strategy Actor — Subplan Emission
+#
+# Demonstrates a strategy actor that uses builtin/plan-subplan to spawn
+# sequential and parallel child plans during the Strategize phase.
+#
+# The actor calls builtin/plan-subplan with the following payload:
+#   - goal:            What the child plan should accomplish
+#   - resource_scopes: Paths or resource IDs the child plan targets
+#   - parallel:        True → SUBPLAN_PARALLEL_SPAWN; False → SUBPLAN_SPAWN
+#   - merge_strategy:  How to merge child-plan results (default: git_three_way)
+#   - max_parallel:    Maximum concurrent subplans (default: 5)
+#   - dependencies:    Sibling subplan ULIDs this plan depends on
+
+name: local/strategy_with_subplan
+type: llm
+description: >
+  Strategy actor that decomposes a large plan into coordinated subplans
+  using the builtin/plan-subplan tool.
+version: "1.0"
+
+model: gpt-4-turbo
+context_view: strategist
+
+system_prompt: |
+  You are a strategy actor responsible for decomposing complex plans into
+  child subplans. Use the builtin/plan-subplan tool to emit subplan decisions.
+
+  For sequential work (one subplan at a time):
+    Call plan-subplan with parallel=false.
+
+  For parallel work (multiple subplans at once):
+    Call plan-subplan with parallel=true and list any dependencies.
+
+  Always include at least one resource_scope or project_ref.
+
+tools:
+  - builtin/plan-subplan
+
+memory:
+  enabled: false
+
+# ---------------------------------------------------------------------------
+# Example: Sequential subplan payload
+#
+# {
+#   "goal": "Refactor the authentication module",
+#   "plan_id": "01ARZ3NDEKTSV4RRFFQ69G5FAV",
+#   "sequence_number": 1,
+#   "resource_scopes": ["src/auth/", "tests/auth/"],
+#   "parallel": false,
+#   "merge_strategy": "git_three_way"
+# }
+#
+# Example: Parallel subplan payload with dependencies
+#
+# {
+#   "goal": "Run full test suite",
+#   "plan_id": "01ARZ3NDEKTSV4RRFFQ69G5FAV",
+#   "sequence_number": 2,
+#   "resource_scopes": ["tests/"],
+#   "parallel": true,
+#   "max_parallel": 3,
+#   "dependencies": ["01ARZ3NDEKTSV4RRFFQ69G5FAW"],
+#   "merge_strategy": "fail_on_conflict"
+# }
+# ---------------------------------------------------------------------------

features/plan_subplan_tool.feature

@@ -0,0 +1,139 @@
+Feature: plan_subplan builtin tool — decision emission
+  As a strategy actor
+  I want to call the builtin/plan-subplan tool
+  So that SUBPLAN_SPAWN or SUBPLAN_PARALLEL_SPAWN decisions are emitted
+  with validated payloads, correct defaults, and rich rationale text
+
+  # ---------------------------------------------------------------------------
+  # Payload validation
+  # ---------------------------------------------------------------------------
+
+  Scenario: Serial subplan with resource_scopes succeeds
+    Given a plan_subplan tool handler
+    When I invoke plan_subplan with goal "Refactor auth module" and resource_scopes ["src/auth"]
+    Then the plan_subplan result should succeed
+    And the plan_subplan result decision_type should be "subplan_spawn"
+
+  Scenario: Serial subplan with project_ref succeeds
+    Given a plan_subplan tool handler
+    When I invoke plan_subplan with goal "Migrate database" and project_ref "projects/db"
+    Then the plan_subplan result should succeed
+    And the plan_subplan result decision_type should be "subplan_spawn"
+
+  Scenario: Parallel subplan with dependencies succeeds
+    Given a plan_subplan tool handler
+    When I invoke parallel plan_subplan with dependencies ["01ARZ3NDEKTSV4RRFFQ69G5FAV"]
+    Then the plan_subplan result should succeed
+    And the plan_subplan result decision_type should be "subplan_parallel_spawn"
+
+  Scenario: Missing resource_scopes and project_ref raises error
+    Given a plan_subplan tool handler
+    When I invoke plan_subplan with goal "Bad call" and no resource scope or project ref
+    Then the plan_subplan result should fail
+    And the plan_subplan error should mention "resource_scopes"
+
+  Scenario: Empty goal raises error
+    Given a plan_subplan tool handler
+    When I invoke plan_subplan with empty goal and resource_scopes ["src/"]
+    Then the plan_subplan result should fail
+    And the plan_subplan error should mention "goal"
+
+  Scenario: max_parallel below minimum raises error
+    Given a plan_subplan tool handler
+    When I invoke parallel plan_subplan with max_parallel 0
+    Then the plan_subplan result should fail
+    And the plan_subplan error should mention "max_parallel"
+
+  Scenario: max_parallel above maximum raises error
+    Given a plan_subplan tool handler
+    When I invoke parallel plan_subplan with max_parallel 51
+    Then the plan_subplan result should fail
+    And the plan_subplan error should mention "max_parallel"
+
+  Scenario: Invalid merge_strategy raises error
+    Given a plan_subplan tool handler
+    When I invoke plan_subplan with invalid merge_strategy "merge_magic"
+    Then the plan_subplan result should fail
+    And the plan_subplan error should mention "merge_strategy"
+
+  Scenario: Invalid context_view raises error
+    Given a plan_subplan tool handler
+    When I invoke plan_subplan with invalid context_view "omniscient"
+    Then the plan_subplan result should fail
+    And the plan_subplan error should mention "context_view"
+
+  # ---------------------------------------------------------------------------
+  # Default values
+  # ---------------------------------------------------------------------------
+
+  Scenario: merge_strategy defaults to git_three_way
+    Given a plan_subplan tool handler
+    When I invoke plan_subplan with goal "Deploy frontend" and resource_scopes ["src/ui"]
+    Then the plan_subplan result output merge_strategy should be "git_three_way"
+
+  Scenario: max_parallel defaults to 5
+    Given a plan_subplan tool handler
+    When I invoke parallel plan_subplan with no max_parallel override
+    Then the plan_subplan result output max_parallel should be 5
+
+  Scenario: parallel defaults to False for serial spawn
+    Given a plan_subplan tool handler
+    When I invoke plan_subplan with goal "Clean logs" and resource_scopes ["logs/"]
+    Then the plan_subplan result output parallel should be false
+
+  Scenario: dependencies defaults to empty list
+    Given a plan_subplan tool handler
+    When I invoke plan_subplan with goal "Build docs" and resource_scopes ["docs/"]
+    Then the plan_subplan result output dependencies should be empty
+
+  # ---------------------------------------------------------------------------
+  # Rationale text
+  # ---------------------------------------------------------------------------
+
+  Scenario: Rationale is auto-generated for serial subplan
+    Given a plan_subplan tool handler
+    When I invoke plan_subplan with goal "Refactor models" and resource_scopes ["src/models"]
+    Then the plan_subplan result rationale should mention the goal
+
+  Scenario: Rationale for parallel subplan mentions parallel execution
+    Given a plan_subplan tool handler
+    When I invoke parallel plan_subplan with dependencies []
+    Then the plan_subplan result rationale should mention "parallel"
+
+  # ---------------------------------------------------------------------------
+  # Context view override
+  # ---------------------------------------------------------------------------
+
+  Scenario: Context view override is stored in output
+    Given a plan_subplan tool handler
+    When I invoke plan_subplan with context_view "executor" and resource_scopes ["src/"]
+    Then the plan_subplan result output context_view should be "executor"
+
+  Scenario: parent_decision_id is included in the decision when provided
+    Given a plan_subplan tool handler
+    When I invoke plan_subplan with a parent_decision_id and resource_scopes ["src/"]
+    Then the plan_subplan result should succeed
+    And the plan_subplan result decision_type should be "subplan_spawn"
+
+  # ---------------------------------------------------------------------------
+  # Decision service integration
+  # ---------------------------------------------------------------------------
+
+  Scenario: Tool records decision via injected decision service
+    Given a plan_subplan tool handler with a stub decision service
+    When I invoke plan_subplan with goal "Run tests" and resource_scopes ["tests/"]
+    Then the stub decision service should have recorded 1 decision
+    And the recorded decision type should be "subplan_spawn"
+
+  # ---------------------------------------------------------------------------
+  # Registration and execution via ToolRunner
+  # ---------------------------------------------------------------------------
+
+  Scenario: Tool registers successfully into ToolRegistry
+    When I register the plan_subplan tool in a ToolRegistry
+    Then the registry should contain "builtin/plan-subplan"
+
+  Scenario: Tool executes via ToolRunner and returns success
+    Given a ToolRunner with the plan_subplan tool registered
+    When I run plan_subplan via the ToolRunner with valid inputs
+    Then the ToolRunner result should succeed

features/steps/plan_subplan_tool_steps.py

@@ -0,0 +1,320 @@
+"""Step definitions for plan_subplan_tool.feature.
+
+Tests the builtin/plan-subplan tool: payload validation, defaults,
+decision emission, rationale generation, and ToolRunner integration.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from behave import given, then, when  # type: ignore[import-untyped]
+from behave.runner import Context  # type: ignore[import-untyped]
+from ulid import ULID
+
+from cleveragents.domain.models.core.decision import Decision
+from cleveragents.tool.builtins.subplan_tool import (
+    PLAN_SUBPLAN_SPEC,
+    make_plan_subplan_spec,
+)
+from cleveragents.tool.registry import ToolRegistry
+from cleveragents.tool.runner import ToolRunner
+
+# ---------------------------------------------------------------------------
+# Constants
+# ---------------------------------------------------------------------------
+
+_PLAN_ID = str(ULID())
+_SEQ = 0
+
+
+# ---------------------------------------------------------------------------
+# Stub decision service
+# ---------------------------------------------------------------------------
+
+
+class _StubDecisionService:
+    """In-memory stub that records decisions without a database."""
+
+    def __init__(self) -> None:
+        self.recorded: list[Decision] = []
+
+    def record_decision(self, decision: Decision) -> Decision:
+        self.recorded.append(decision)
+        return decision
+
+
+# ---------------------------------------------------------------------------
+# Helper: build minimal valid inputs
+# ---------------------------------------------------------------------------
+
+
+def _inputs(
+    goal: str = "Implement feature X",
+    resource_scopes: list[str] | None = None,
+    project_ref: str | None = None,
+    parallel: bool = False,
+    merge_strategy: str | None = None,
+    max_parallel: int | None = None,
+    dependencies: list[str] | None = None,
+    context_view: str | None = None,
+) -> dict[str, Any]:
+    payload: dict[str, Any] = {
+        "goal": goal,
+        "plan_id": _PLAN_ID,
+        "sequence_number": _SEQ,
+        "parallel": parallel,
+    }
+    if resource_scopes is not None:
+        payload["resource_scopes"] = resource_scopes
+    if project_ref is not None:
+        payload["project_ref"] = project_ref
+    if merge_strategy is not None:
+        payload["merge_strategy"] = merge_strategy
+    if max_parallel is not None:
+        payload["max_parallel"] = max_parallel
+    if dependencies is not None:
+        payload["dependencies"] = dependencies
+    if context_view is not None:
+        payload["context_view"] = context_view
+    return payload
+
+
+# ---------------------------------------------------------------------------
+# Given
+# ---------------------------------------------------------------------------
+
+
+@given("a plan_subplan tool handler")
+def step_given_handler(ctx: Context) -> None:
+    ctx.subplan_handler = PLAN_SUBPLAN_SPEC.handler
+
+
+@given("a plan_subplan tool handler with a stub decision service")
+def step_given_handler_with_service(ctx: Context) -> None:
+    ctx.stub_service = _StubDecisionService()
+    spec = make_plan_subplan_spec(decision_service=ctx.stub_service)
+    ctx.subplan_handler = spec.handler
+
+
+@given("a ToolRunner with the plan_subplan tool registered")
+def step_given_runner(ctx: Context) -> None:
+    registry = ToolRegistry()
+    registry.register(PLAN_SUBPLAN_SPEC)
+    ctx.runner = ToolRunner(registry)
+
+
+# ---------------------------------------------------------------------------
+# When
+# ---------------------------------------------------------------------------
+
+
+@when('I invoke plan_subplan with goal "{goal}" and resource_scopes {scopes}')
+def step_when_with_scopes(ctx: Context, goal: str, scopes: str) -> None:
+    ctx.result = ctx.subplan_handler(
+        _inputs(goal=goal, resource_scopes=json.loads(scopes))
+    )
+
+
+@when('I invoke plan_subplan with goal "{goal}" and project_ref "{ref}"')
+def step_when_with_project_ref(ctx: Context, goal: str, ref: str) -> None:
+    ctx.result = ctx.subplan_handler(_inputs(goal=goal, project_ref=ref))
+
+
+@when("I invoke parallel plan_subplan with dependencies {deps}")
+def step_when_parallel_with_deps(ctx: Context, deps: str) -> None:
+    ctx.result = ctx.subplan_handler(
+        _inputs(
+            goal="Parallel work",
+            resource_scopes=["src/"],
+            parallel=True,
+            dependencies=json.loads(deps),
+        )
+    )
+
+
+@when('I invoke plan_subplan with goal "{goal}" and no resource scope or project ref')
+def step_when_no_scope(ctx: Context, goal: str) -> None:
+    ctx.result = ctx.subplan_handler(
+        {"goal": goal, "plan_id": _PLAN_ID, "sequence_number": 0}
+    )
+
+
+@when("I invoke plan_subplan with empty goal and resource_scopes {scopes}")
+def step_when_empty_goal(ctx: Context, scopes: str) -> None:
+    ctx.result = ctx.subplan_handler(
+        _inputs(goal="", resource_scopes=json.loads(scopes))
+    )
+
+
+@when("I invoke parallel plan_subplan with max_parallel {n:d}")
+def step_when_bad_max_parallel(ctx: Context, n: int) -> None:
+    ctx.result = ctx.subplan_handler(
+        _inputs(resource_scopes=["src/"], parallel=True, max_parallel=n)
+    )
+
+
+@when('I invoke plan_subplan with invalid merge_strategy "{strategy}"')
+def step_when_bad_strategy(ctx: Context, strategy: str) -> None:
+    ctx.result = ctx.subplan_handler(
+        _inputs(resource_scopes=["src/"], merge_strategy=strategy)
+    )
+
+
+@when('I invoke plan_subplan with invalid context_view "{view}"')
+def step_when_bad_context_view(ctx: Context, view: str) -> None:
+    ctx.result = ctx.subplan_handler(
+        _inputs(resource_scopes=["src/"], context_view=view)
+    )
+
+
+@when("I invoke plan_subplan with a parent_decision_id and resource_scopes {scopes}")
+def step_when_with_parent_decision(ctx: Context, scopes: str) -> None:
+    parent_id = "01ARZ3NDEKTSV4RRFFQ69G5FAV"
+    inp = _inputs(resource_scopes=json.loads(scopes))
+    inp["parent_decision_id"] = parent_id
+    ctx.result = ctx.subplan_handler(inp)
+
+
+@when("I invoke parallel plan_subplan with no max_parallel override")
+def step_when_default_max_parallel(ctx: Context) -> None:
+    ctx.result = ctx.subplan_handler(_inputs(resource_scopes=["src/"], parallel=True))
+
+
+@when('I invoke plan_subplan with context_view "{view}" and resource_scopes {scopes}')
+def step_when_context_view(ctx: Context, view: str, scopes: str) -> None:
+    ctx.result = ctx.subplan_handler(
+        _inputs(resource_scopes=json.loads(scopes), context_view=view)
+    )
+
+
+@when("I register the plan_subplan tool in a ToolRegistry")
+def step_when_register(ctx: Context) -> None:
+    ctx.registry = ToolRegistry()
+    ctx.registry.register(PLAN_SUBPLAN_SPEC)
+
+
+@when("I run plan_subplan via the ToolRunner with valid inputs")
+def step_when_run_via_runner(ctx: Context) -> None:
+    ctx.runner_result = ctx.runner.execute(
+        "builtin/plan-subplan",
+        _inputs(goal="Runner test", resource_scopes=["src/"]),
+    )
+
+
+# ---------------------------------------------------------------------------
+# Then
+# ---------------------------------------------------------------------------
+
+
+@then("the plan_subplan result should succeed")
+def step_then_success(ctx: Context) -> None:
+    r = ctx.result
+    assert isinstance(r, dict), f"Expected dict, got {type(r)}"
+    assert r.get("success") is True, f"Expected success=True, got: {r}"
+
+
+@then("the plan_subplan result should fail")
+def step_then_fail(ctx: Context) -> None:
+    r = ctx.result
+    assert isinstance(r, dict), f"Expected dict, got {type(r)}"
+    assert r.get("success") is False, f"Expected success=False, got: {r}"
+
+
+@then('the plan_subplan result decision_type should be "{dtype}"')
+def step_then_dtype(ctx: Context, dtype: str) -> None:
+    r = ctx.result
+    assert r.get("decision_type") == dtype, (
+        f"Expected decision_type={dtype!r}, got: {r.get('decision_type')!r}"
+    )
+
+
+@then('the plan_subplan error should mention "{keyword}"')
+def step_then_error_keyword(ctx: Context, keyword: str) -> None:
+    err = (r := ctx.result).get("error") or ""
+    assert keyword.lower() in err.lower(), (
+        f"Expected error to mention {keyword!r} in {r}"
+    )
+
+
+@then('the plan_subplan result output merge_strategy should be "{value}"')
+def step_then_merge_strategy(ctx: Context, value: str) -> None:
+    r = ctx.result
+    assert r.get("merge_strategy") == value, (
+        f"Expected merge_strategy={value!r}, got: {r.get('merge_strategy')!r}"
+    )
+
+
+@then("the plan_subplan result output max_parallel should be {n:d}")
+def step_then_max_parallel(ctx: Context, n: int) -> None:
+    r = ctx.result
+    assert r.get("max_parallel") == n, (
+        f"Expected max_parallel={n}, got: {r.get('max_parallel')!r}"
+    )
+
+
+@then("the plan_subplan result output parallel should be false")
+def step_then_not_parallel(ctx: Context) -> None:
+    r = ctx.result
+    assert r.get("parallel") is False, (
+        f"Expected parallel=False, got: {r.get('parallel')!r}"
+    )
+
+
+@then("the plan_subplan result output dependencies should be empty")
+def step_then_empty_deps(ctx: Context) -> None:
+    r = ctx.result
+    assert r.get("dependencies") == [], f"Expected [], got: {r.get('dependencies')!r}"
+
+
+@then("the plan_subplan result rationale should mention the goal")
+def step_then_rationale_goal(ctx: Context) -> None:
+    r = ctx.result
+    rationale = r.get("rationale") or ""
+    assert rationale, "Expected non-empty rationale"
+
+
+@then('the plan_subplan result rationale should mention "{keyword}"')
+def step_then_rationale_keyword(ctx: Context, keyword: str) -> None:
+    rationale = (ctx.result.get("rationale") or "").lower()
+    assert keyword.lower() in rationale, (
+        f"Expected rationale to mention {keyword!r}, got: {rationale!r}"
+    )
+
+
+@then('the plan_subplan result output context_view should be "{view}"')
+def step_then_context_view(ctx: Context, view: str) -> None:
+    r = ctx.result
+    assert r.get("context_view") == view, (
+        f"Expected context_view={view!r}, got: {r.get('context_view')!r}"
+    )
+
+
+@then("the stub decision service should have recorded {n:d} decision")
+def step_then_service_recorded(ctx: Context, n: int) -> None:
+    stub: _StubDecisionService = ctx.stub_service
+    assert len(stub.recorded) == n, (
+        f"Expected {n} recorded decisions, got {len(stub.recorded)}"
+    )
+
+
+@then('the recorded decision type should be "{dtype}"')
+def step_then_recorded_dtype(ctx: Context, dtype: str) -> None:
+    stub: _StubDecisionService = ctx.stub_service
+    assert stub.recorded, "No decisions were recorded"
+    assert str(stub.recorded[0].decision_type) == dtype, (
+        f"Expected {dtype!r}, got {stub.recorded[0].decision_type!r}"
+    )
+
+
+@then('the registry should contain "{name}"')
+def step_then_registry_contains(ctx: Context, name: str) -> None:
+    spec = ctx.registry.get(name)
+    assert spec is not None, f"Expected '{name}' in registry"
+
+
+@then("the ToolRunner result should succeed")
+def step_then_runner_success(ctx: Context) -> None:
+    result = ctx.runner_result
+    assert result.success, f"Expected success=True, error={result.error}"

robot/helper_plan_subplan_tool.py

@@ -0,0 +1,190 @@
+"""Helper script for Robot Framework plan_subplan tool smoke tests.
+
+Each command prints a sentinel string on success so the Robot suite can
+assert with ``Should Contain``.
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from typing import Any
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
+
+from cleveragents.domain.models.core.decision import Decision, DecisionType
+from cleveragents.tool.builtins import register_subplan_tool
+from cleveragents.tool.builtins.subplan_tool import (
+    PLAN_SUBPLAN_SPEC,
+    SubplanPayload,
+    make_plan_subplan_spec,
+)
+from cleveragents.tool.registry import ToolRegistry
+from cleveragents.tool.runner import ToolRunner
+
+_PLAN_ID = "01ARZ3NDEKTSV4RRFFQ69G5FAV"
+
+
+# ---------------------------------------------------------------------------
+# Stub decision service
+# ---------------------------------------------------------------------------
+
+
+class _StubService:
+    def __init__(self) -> None:
+        self.recorded: list[Decision] = []
+
+    def record_decision(self, decision: Decision) -> Decision:
+        self.recorded.append(decision)
+        return decision
+
+
+# ---------------------------------------------------------------------------
+# Helper to build minimal valid inputs
+# ---------------------------------------------------------------------------
+
+
+def _base(
+    goal: str = "Refactor module",
+    resource_scopes: list[str] | None = None,
+    parallel: bool = False,
+    **extra: Any,
+) -> dict[str, Any]:
+    return {
+        "goal": goal,
+        "plan_id": _PLAN_ID,
+        "sequence_number": 0,
+        "resource_scopes": resource_scopes or ["src/"],
+        "parallel": parallel,
+        **extra,
+    }
+
+
+# ---------------------------------------------------------------------------
+# Test functions
+# ---------------------------------------------------------------------------
+
+
+def test_serial_spawn() -> None:
+    result = PLAN_SUBPLAN_SPEC.handler(
+        _base(goal="Refactor auth", resource_scopes=["src/auth/"])
+    )
+    assert result["success"] is True
+    assert result["decision_type"] == "subplan_spawn"
+    assert result["parallel"] is False
+    print("serial-spawn-ok")
+
+
+def test_parallel_spawn() -> None:
+    result = PLAN_SUBPLAN_SPEC.handler(
+        _base(
+            goal="Run tests in parallel",
+            resource_scopes=["tests/"],
+            parallel=True,
+            max_parallel=3,
+            dependencies=[],
+        )
+    )
+    assert result["success"] is True
+    assert result["decision_type"] == "subplan_parallel_spawn"
+    assert result["parallel"] is True
+    print("parallel-spawn-ok")
+
+
+def test_validation_error() -> None:
+    result = PLAN_SUBPLAN_SPEC.handler(
+        {"goal": "Bad", "plan_id": _PLAN_ID, "sequence_number": 0}
+    )
+    assert result["success"] is False
+    err = result["error"].lower()
+    assert "resource_scopes" in err or "project_ref" in err
+    print("validation-error-ok")
+
+
+def test_defaults() -> None:
+    result = PLAN_SUBPLAN_SPEC.handler(_base())
+    assert result["merge_strategy"] == "git_three_way"
+    assert result["max_parallel"] == 5
+    assert result["dependencies"] == []
+    print("defaults-ok")
+
+
+def test_decision_service_injection() -> None:
+    stub = _StubService()
+    spec = make_plan_subplan_spec(decision_service=stub)
+    result = spec.handler(_base(goal="Run linting", resource_scopes=["src/"]))
+    assert result["success"] is True
+    assert len(stub.recorded) == 1
+    assert stub.recorded[0].decision_type == DecisionType.SUBPLAN_SPAWN
+    print("service-injection-ok")
+
+
+def test_registration() -> None:
+    registry = ToolRegistry()
+    register_subplan_tool(registry)
+    spec = registry.get("builtin/plan-subplan")
+    assert spec is not None
+    assert spec.name == "builtin/plan-subplan"
+    print("registration-ok")
+
+
+def test_runner_execution() -> None:
+    registry = ToolRegistry()
+    register_subplan_tool(registry)
+    runner = ToolRunner(registry)
+    result = runner.execute(
+        "builtin/plan-subplan",
+        _base(goal="Via runner", resource_scopes=["src/"]),
+    )
+    assert result.success, f"Runner failed: {result.error}"
+    assert result.output["decision_type"] == "subplan_spawn"
+    print("runner-execution-ok")
+
+
+def test_rationale_content() -> None:
+    result = PLAN_SUBPLAN_SPEC.handler(
+        _base(goal="Migrate to PostgreSQL", resource_scopes=["db/"])
+    )
+    assert result["success"] is True
+    rationale = result["rationale"]
+    assert "Migrate to PostgreSQL" in rationale
+    assert "sequential" in rationale
+    print("rationale-ok")
+
+
+def test_payload_model_direct() -> None:
+    payload = SubplanPayload(
+        goal="Direct model test",
+        plan_id=_PLAN_ID,
+        sequence_number=1,
+        resource_scopes=["src/"],
+        parallel=True,
+        max_parallel=10,
+    )
+    assert payload.merge_strategy.value == "git_three_way"
+    assert payload.max_parallel == 10
+    assert payload.parallel is True
+    print("payload-model-ok")
+
+
+# ---------------------------------------------------------------------------
+# Dispatch
+# ---------------------------------------------------------------------------
+
+_COMMANDS: dict[str, Any] = {
+    "serial": test_serial_spawn,
+    "parallel": test_parallel_spawn,
+    "validation": test_validation_error,
+    "defaults": test_defaults,
+    "service": test_decision_service_injection,
+    "registration": test_registration,
+    "runner": test_runner_execution,
+    "rationale": test_rationale_content,
+    "payload": test_payload_model_direct,
+}
+
+if __name__ == "__main__":
+    if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
+        print(f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>", file=sys.stderr)
+        sys.exit(1)
+    _COMMANDS[sys.argv[1]]()

robot/plan_subplan_tool.robot

@@ -0,0 +1,63 @@
+*** Settings ***
+Documentation    Smoke tests for builtin/plan-subplan actor tool
+Resource         ${CURDIR}/common.resource
+Suite Setup      Setup Test Environment
+Suite Teardown   Cleanup Test Environment
+
+*** Variables ***
+${HELPER_SCRIPT}    robot/helper_plan_subplan_tool.py
+
+*** Test Cases ***
+Serial Subplan Spawn
+    [Documentation]    Emit a SUBPLAN_SPAWN decision for a serial subplan
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    serial    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    serial-spawn-ok
+
+Parallel Subplan Spawn
+    [Documentation]    Emit a SUBPLAN_PARALLEL_SPAWN decision for a parallel subplan
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    parallel    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    parallel-spawn-ok
+
+Validation Error Without Scope
+    [Documentation]    Tool returns error when no resource_scopes or project_ref provided
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    validation    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    validation-error-ok
+
+Default Values Applied
+    [Documentation]    merge_strategy, max_parallel, and dependencies use sensible defaults
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    defaults    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    defaults-ok
+
+Decision Service Injection
+    [Documentation]    Injected DecisionService receives the emitted Decision object
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    service    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    service-injection-ok
+
+Tool Registration In Registry
+    [Documentation]    register_subplan_tool adds builtin/plan-subplan to ToolRegistry
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    registration    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    registration-ok
+
+ToolRunner Execution
+    [Documentation]    ToolRunner executes builtin/plan-subplan and returns success
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    runner    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    runner-execution-ok
+
+Rationale Text Contains Goal
+    [Documentation]    Rationale text includes goal and execution mode
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    rationale    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    rationale-ok
+
+SubplanPayload Model Direct Validation
+    [Documentation]    SubplanPayload can be constructed directly for testing
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    payload    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    payload-model-ok

src/cleveragents/tool/builtins/__init__.py

@@ -1,7 +1,8 @@
 """Built-in tool registration helpers.
 
-Provides convenience functions to register all built-in file and git tools
-into a ``ToolRegistry``, optionally wrapped with changeset capture.
+Provides convenience functions to register all built-in file, git, and
+subplan tools into a ``ToolRegistry``, optionally wrapped with changeset
+capture.
 """
 
 from __future__ import annotations
@@ -29,6 +30,12 @@ from cleveragents.tool.builtins.git_tools import (
     GIT_STATUS_SPEC,
     validate_repo_path,
 )
+from cleveragents.tool.builtins.subplan_tool import (
+    ALL_SUBPLAN_TOOLS,
+    PLAN_SUBPLAN_SPEC,
+    SubplanPayload,
+    make_plan_subplan_spec,
+)
 from cleveragents.tool.registry import ToolRegistry
 
 
@@ -58,9 +65,16 @@ def register_git_tools(registry: ToolRegistry) -> None:
         registry.register(spec)
 
 
+def register_subplan_tool(registry: ToolRegistry) -> None:
+    """Register the ``builtin/plan-subplan`` tool into *registry*."""
+    for spec in ALL_SUBPLAN_TOOLS:
+        registry.register(spec)
+
+
 __all__ = [
     "ALL_FILE_TOOLS",
     "ALL_GIT_TOOLS",
+    "ALL_SUBPLAN_TOOLS",
     "FILE_DELETE_SPEC",
     "FILE_EDIT_SPEC",
     "FILE_LIST_SPEC",
@@ -71,12 +85,16 @@ __all__ = [
     "GIT_DIFF_SPEC",
     "GIT_LOG_SPEC",
     "GIT_STATUS_SPEC",
+    "PLAN_SUBPLAN_SPEC",
     "ChangeSet",
     "ChangeSetCapture",
     "ChangeSetEntry",
+    "SubplanPayload",
+    "make_plan_subplan_spec",
     "register_file_tools",
     "register_file_tools_with_changeset",
     "register_git_tools",
+    "register_subplan_tool",
     "validate_path",
     "validate_repo_path",
 ]

src/cleveragents/tool/builtins/subplan_tool.py

@@ -0,0 +1,395 @@
+"""Built-in plan_subplan tool for CleverAgents strategy actors.
+
+Provides the ``builtin/plan-subplan`` tool which strategy actors call
+to emit ``SUBPLAN_SPAWN`` or ``SUBPLAN_PARALLEL_SPAWN`` decisions when
+decomposing a plan into coordinated child plans.
+
+## Tool Overview
+
+| Attribute      | Value                                                 |
+|----------------|-------------------------------------------------------|
+| Tool name      | ``builtin/plan-subplan``                              |
+| Capability     | ``writes=True``, ``checkpointable=False``             |
+| Decision types | ``subplan_spawn`` / ``subplan_parallel_spawn``        |
+| Namespace      | ``builtin/``                                          |
+
+## Usage
+
+```python
+from cleveragents.tool.builtins import register_subplan_tool
+from cleveragents.tool.registry import ToolRegistry
+
+registry = ToolRegistry()
+register_subplan_tool(registry)
+```
+
+With an injected ``DecisionService`` for persistence:
+
+```python
+from cleveragents.tool.builtins.subplan_tool import make_plan_subplan_spec
+
+spec = make_plan_subplan_spec(decision_service=my_service)
+registry.register(spec)
+```
+
+## Input Schema
+
+Required fields: ``goal``, ``plan_id``, ``sequence_number``.
+
+Optional fields and defaults:
+
+- ``resource_scopes`` (list[str], default ``[]``) — paths the subplan targets
+- ``project_ref`` (str, default ``None``) — alternative to resource_scopes
+- ``parallel`` (bool, default ``False``) — emit parallel-spawn decision
+- ``merge_strategy`` (str, default ``git_three_way``) — merge behaviour
+- ``max_parallel`` (int 1-50, default ``5``) — max concurrent subplans
+- ``dependencies`` (list[str], default ``[]``) — sibling ULIDs
+- ``context_view`` (str, default ``None``) — actor context-view override
+- ``parent_decision_id`` (str, default ``None``) — parent decision ULID
+
+*At least one of ``resource_scopes`` (non-empty) or ``project_ref`` is required.
+
+Based on:
+    - docs/specification.md §Child Plan Spawning Mechanism
+    - Forgejo issue #198 — feat(actor): add plan_subplan tool and decision emission
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
+
+from cleveragents.domain.models.core.decision import Decision, DecisionType
+from cleveragents.domain.models.core.plan import SubplanMergeStrategy
+from cleveragents.domain.models.core.tool import ToolCapability
+from cleveragents.tool.runtime import ToolSpec
+
+if TYPE_CHECKING:
+    pass
+
+# ---------------------------------------------------------------------------
+# Constants
+# ---------------------------------------------------------------------------
+
+TOOL_NAME = "builtin/plan-subplan"
+
+_DEFAULT_MERGE_STRATEGY = SubplanMergeStrategy.GIT_THREE_WAY
+_DEFAULT_MAX_PARALLEL = 5
+_VALID_CONTEXT_VIEWS: frozenset[str] = frozenset(
+    {"strategist", "executor", "reviewer", "full"}
+)
+
+
+# ---------------------------------------------------------------------------
+# Protocol for optional decision service injection
+# ---------------------------------------------------------------------------
+
+
+@runtime_checkable
+class _DecisionRecorder(Protocol):
+    """Protocol for recording decisions (subset of DecisionService API)."""
+
+    def record_decision(self, decision: Decision) -> Decision: ...
+
+
+# ---------------------------------------------------------------------------
+# Input payload model
+# ---------------------------------------------------------------------------
+
+
+class SubplanPayload(BaseModel):
+    """Validated and defaulted inputs for the plan_subplan tool.
+
+    Enforces all acceptance-criteria constraints:
+    - At least one of ``resource_scopes`` or ``project_ref`` must be provided.
+    - ``max_parallel`` must be in [1, 50].
+    - ``merge_strategy`` must be a valid ``SubplanMergeStrategy`` value.
+    - ``goal`` must be non-empty.
+    """
+
+    model_config = ConfigDict(frozen=True, str_strip_whitespace=True)
+
+    goal: str = Field(..., min_length=1, description="Goal of the child subplan.")
+    plan_id: str = Field(..., min_length=1, description="ULID of the spawning plan.")
+    sequence_number: int = Field(
+        ..., ge=0, description="Monotonic sequence number within the plan."
+    )
+    resource_scopes: list[str] = Field(
+        default_factory=list,
+        description="Paths or resource IDs the subplan targets.",
+    )
+    project_ref: str | None = Field(
+        default=None,
+        description="Project reference (alternative to resource_scopes).",
+    )
+    parallel: bool = Field(
+        default=False,
+        description="Emit SUBPLAN_PARALLEL_SPAWN instead of SUBPLAN_SPAWN.",
+    )
+    merge_strategy: SubplanMergeStrategy = Field(
+        default=_DEFAULT_MERGE_STRATEGY,
+        description="How to merge child-plan results into the parent.",
+    )
+    max_parallel: int = Field(
+        default=_DEFAULT_MAX_PARALLEL,
+        ge=1,
+        le=50,
+        description="Max concurrent subplans (only used when parallel=True).",
+    )
+    dependencies: list[str] = Field(
+        default_factory=list,
+        description="ULIDs of sibling subplans this one depends on.",
+    )
+    context_view: str | None = Field(
+        default=None,
+        description="Actor context-view override for the child plan.",
+    )
+    parent_decision_id: str | None = Field(
+        default=None,
+        description="ULID of the parent decision node.",
+    )
+
+    @field_validator("context_view")
+    @classmethod
+    def _validate_context_view(cls, v: str | None) -> str | None:
+        if v is not None and v not in _VALID_CONTEXT_VIEWS:
+            raise ValueError(
+                f"context_view must be one of {sorted(_VALID_CONTEXT_VIEWS)}, got {v!r}"
+            )
+        return v
+
+    @model_validator(mode="after")
+    def _require_scope_or_ref(self) -> SubplanPayload:
+        has_scopes = bool(self.resource_scopes)
+        has_ref = bool(self.project_ref and self.project_ref.strip())
+        if not has_scopes and not has_ref:
+            raise ValueError(
+                "At least one of 'resource_scopes' (non-empty) or "
+                "'project_ref' must be provided."
+            )
+        return self
+
+
+# ---------------------------------------------------------------------------
+# Rationale builder
+# ---------------------------------------------------------------------------
+
+
+def _build_rationale(payload: SubplanPayload) -> str:
+    """Build a human-readable rationale explaining why the subplan was spawned.
+
+    The text is surfaced in ``plan explain`` output and stored in the
+    Decision record for audit purposes.
+    """
+    scope_desc = (
+        ", ".join(payload.resource_scopes)
+        if payload.resource_scopes
+        else payload.project_ref or "(unspecified)"
+    )
+    mode = "parallel" if payload.parallel else "sequential"
+    lines = [
+        f"Spawning {mode} subplan to accomplish: {payload.goal}",
+        f"Targeted resources: {scope_desc}",
+        f"Merge strategy: {payload.merge_strategy.value}",
+    ]
+    if payload.parallel and payload.dependencies:
+        dep_str = ", ".join(payload.dependencies)
+        lines.append(f"Depends on subplans: {dep_str}")
+    if payload.context_view:
+        lines.append(f"Context view override: {payload.context_view}")
+    return "\n".join(lines)
+
+
+# ---------------------------------------------------------------------------
+# Tool handler factory
+# ---------------------------------------------------------------------------
+
+
+def _make_handler(
+    decision_service: _DecisionRecorder | None = None,
+) -> Callable[[dict[str, Any]], dict[str, Any]]:
+    """Return a tool handler optionally bound to a *decision_service*.
+
+    When *decision_service* is ``None`` the handler validates and builds
+    the decision but does not persist it.  Callers that need persistence
+    (e.g. the full runtime) should inject a ``DecisionService`` instance.
+    """
+
+    def handler(inputs: dict[str, Any]) -> dict[str, Any]:
+        # ------------------------------------------------------------------
+        # Validate and apply defaults via SubplanPayload
+        # ------------------------------------------------------------------
+        try:
+            payload = SubplanPayload(**inputs)
+        except Exception as exc:
+            return {"success": False, "error": str(exc)}
+
+        # ------------------------------------------------------------------
+        # Determine decision type
+        # ------------------------------------------------------------------
+        decision_type = (
+            DecisionType.SUBPLAN_PARALLEL_SPAWN
+            if payload.parallel
+            else DecisionType.SUBPLAN_SPAWN
+        )
+
+        # ------------------------------------------------------------------
+        # Build rationale
+        # ------------------------------------------------------------------
+        rationale = _build_rationale(payload)
+
+        # ------------------------------------------------------------------
+        # Construct the Decision domain object
+        # ------------------------------------------------------------------
+        decision_kwargs: dict[str, Any] = {
+            "plan_id": payload.plan_id,
+            "sequence_number": payload.sequence_number,
+            "decision_type": decision_type,
+            "question": ("Should a subplan be spawned to accomplish this goal?"),
+            "chosen_option": f"Spawn {decision_type.value} — {payload.goal}",
+            "rationale": rationale,
+        }
+        if payload.parent_decision_id:
+            decision_kwargs["parent_decision_id"] = payload.parent_decision_id
+
+        decision = Decision(**decision_kwargs)
+
+        # ------------------------------------------------------------------
+        # Optionally persist via injected service
+        # ------------------------------------------------------------------
+        if decision_service is not None:
+            decision_service.record_decision(decision)
+
+        # ------------------------------------------------------------------
+        # Return result dict (JSON-serialisable)
+        # ------------------------------------------------------------------
+        return {
+            "success": True,
+            "decision_id": decision.decision_id,
+            "decision_type": str(decision_type.value),
+            "goal": payload.goal,
+            "resource_scopes": payload.resource_scopes,
+            "project_ref": payload.project_ref,
+            "parallel": payload.parallel,
+            "merge_strategy": str(payload.merge_strategy.value),
+            "max_parallel": payload.max_parallel,
+            "dependencies": payload.dependencies,
+            "context_view": payload.context_view,
+            "rationale": rationale,
+        }
+
+    return handler
+
+
+# ---------------------------------------------------------------------------
+# Public factory
+# ---------------------------------------------------------------------------
+
+
+def make_plan_subplan_spec(
+    decision_service: _DecisionRecorder | None = None,
+) -> ToolSpec:
+    """Build a ``ToolSpec`` for the ``builtin/plan-subplan`` tool.
+
+    Args:
+        decision_service: Optional service that implements
+            ``record_decision``.  When supplied the handler persists each
+            emitted decision immediately.  When ``None`` the decision is
+            built and returned but not stored.
+
+    Returns:
+        A ready-to-register :class:`~cleveragents.tool.runtime.ToolSpec`.
+    """
+    return ToolSpec(
+        name=TOOL_NAME,
+        description=(
+            "Emit a SUBPLAN_SPAWN (or SUBPLAN_PARALLEL_SPAWN when "
+            "parallel=True) decision for the current strategy actor. "
+            "Validates the subplan payload and records the decision."
+        ),
+        input_schema={
+            "type": "object",
+            "properties": {
+                "goal": {"type": "string", "minLength": 1},
+                "plan_id": {"type": "string", "minLength": 1},
+                "sequence_number": {"type": "integer", "minimum": 0},
+                "resource_scopes": {
+                    "type": "array",
+                    "items": {"type": "string"},
+                    "default": [],
+                },
+                "project_ref": {"type": ["string", "null"]},
+                "parallel": {"type": "boolean", "default": False},
+                "merge_strategy": {
+                    "type": "string",
+                    "enum": [s.value for s in SubplanMergeStrategy],
+                    "default": _DEFAULT_MERGE_STRATEGY.value,
+                },
+                "max_parallel": {
+                    "type": "integer",
+                    "minimum": 1,
+                    "maximum": 50,
+                    "default": _DEFAULT_MAX_PARALLEL,
+                },
+                "dependencies": {
+                    "type": "array",
+                    "items": {"type": "string"},
+                    "default": [],
+                },
+                "context_view": {
+                    "type": ["string", "null"],
+                    "enum": [*list(_VALID_CONTEXT_VIEWS), None],  # type: ignore[list-item]
+                },
+                "parent_decision_id": {"type": ["string", "null"]},
+            },
+            "required": ["goal", "plan_id", "sequence_number"],
+        },
+        output_schema={
+            "type": "object",
+            "properties": {
+                "success": {"type": "boolean"},
+                "decision_id": {"type": "string"},
+                "decision_type": {"type": "string"},
+                "goal": {"type": "string"},
+                "resource_scopes": {"type": "array"},
+                "project_ref": {"type": ["string", "null"]},
+                "parallel": {"type": "boolean"},
+                "merge_strategy": {"type": "string"},
+                "max_parallel": {"type": "integer"},
+                "dependencies": {"type": "array"},
+                "context_view": {"type": ["string", "null"]},
+                "rationale": {"type": "string"},
+            },
+        },
+        capabilities=ToolCapability(
+            read_only=False,
+            writes=True,
+            checkpointable=False,
+            side_effects=["emit_decision"],
+        ),
+        handler=_make_handler(decision_service),
+        source="builtin",
+    )
+
+
+# ---------------------------------------------------------------------------
+# Default spec (no service injection)
+# ---------------------------------------------------------------------------
+
+#: Ready-to-register spec with no decision persistence.
+#: Use :func:`make_plan_subplan_spec` when you need decision recording.
+PLAN_SUBPLAN_SPEC: ToolSpec = make_plan_subplan_spec()
+
+#: All subplan tools (currently one) for bulk registration.
+ALL_SUBPLAN_TOOLS: list[ToolSpec] = [PLAN_SUBPLAN_SPEC]
+
+
+__all__ = [
+    "ALL_SUBPLAN_TOOLS",
+    "PLAN_SUBPLAN_SPEC",
+    "TOOL_NAME",
+    "SubplanPayload",
+    "make_plan_subplan_spec",
+]