feat(estimation): add estimation actor YAML template and role-aware registration validation

- Add `role_hint` and `response_format` support to actor schema.
- Add non-fatal estimation-role compatibility warnings in actor registration CLI flows.
- Add preflight warning path when `estimation` actor is missing `response_format`.
- Add `examples/actors/estimator.yaml` and update actor examples documentation/tests.
- Update integration helper expectations (m1/m2/m3/m6) for missing provider config in local test env.

ISSUES CLOSED: #650
This commit is contained in:
2026-03-16 12:37:52 +00:00
parent ab1fd19bcd
commit 26ad778aee
18 changed files with 369 additions and 14 deletions
+19 -5
View File
@@ -17,11 +17,12 @@ This document provides comprehensive examples of actor YAML configurations, demo
2. [Strategist Actors](#strategist-actors)
3. [Executor Actors](#executor-actors)
4. [Reviewer Actors](#reviewer-actors)
5. [Tool-Only Actors](#tool-only-actors)
6. [Validation-Node Actors](#validation-node-actors)
7. [Graph Workflows](#graph-workflows)
8. [Hierarchical Actor Graphs](#hierarchical-actor-graphs)
9. [Strategy Actor with Subplan Spawning](#strategy-actor-with-subplan-spawning)
5. [Estimation Actors](#estimation-actors)
6. [Tool-Only Actors](#tool-only-actors)
7. [Validation-Node Actors](#validation-node-actors)
8. [Graph Workflows](#graph-workflows)
9. [Hierarchical Actor Graphs](#hierarchical-actor-graphs)
10. [Strategy Actor with Subplan Spawning](#strategy-actor-with-subplan-spawning)
---
@@ -283,6 +284,19 @@ tools:
---
## Estimation Actors
### Structured Estimation Reporter
**File:** `examples/actors/estimator.yaml`
This actor is intended for the optional `estimation_actor` slot on actions. It uses
`context_view: strategist`, declares `role_hint: estimation`, and defines
`response_format` with an `EstimationReport` JSON schema so outputs are consistently
machine-readable.
---
## LLM Actors with Tools
### Strategist with File Access
+62
View File
@@ -0,0 +1,62 @@
# Estimation Actor Example
# Produces structured cost/risk/duration estimates prior to execution.
name: local/estimator
type: llm
description: Estimates implementation effort, risk, and timeline before execution
version: "1.0"
model: gpt-4
role_hint: estimation
context_view: strategist
system_prompt: |
You are an estimation specialist.
Assess the requested work and produce:
- Effort estimate (hours and confidence)
- Duration estimate (best/likely/worst case)
- Risk assessment with severity and mitigation suggestions
- Key assumptions and unknowns
Always return valid JSON matching the provided response_format schema.
response_format:
$schema: "https://json-schema.org/draft/2020-12/schema"
title: EstimationReport
type: object
required:
- cost_hours
- duration_days
- risk_level
- confidence
- assumptions
properties:
cost_hours:
type: number
minimum: 0
duration_days:
type: number
minimum: 0
risk_level:
type: string
enum: [low, medium, high]
confidence:
type: number
minimum: 0
maximum: 1
assumptions:
type: array
items:
type: string
mitigation_suggestions:
type: array
items:
type: string
additionalProperties: false
tools:
- files/read_file
- files/list_directory
skills:
- local/file-ops
+3 -1
View File
@@ -653,9 +653,10 @@ Feature: Actor YAML examples validation
Scenario: Verify all example files exist
Given the examples/actors directory
Then there should be exactly 7 example YAML files
Then there should be exactly 8 example YAML files
And the example files should include "simple_llm.yaml"
And the example files should include "llm_with_tools.yaml"
And the example files should include "estimator.yaml"
And the example files should include "tool_collection.yaml"
And the example files should include "simple_graph.yaml"
And the example files should include "graph_workflow.yaml"
@@ -666,6 +667,7 @@ Feature: Actor YAML examples validation
Given the documentation file "docs/reference/actors_examples.md"
Then the documentation should reference "simple_llm.yaml"
And the documentation should reference "llm_with_tools.yaml"
And the documentation should reference "estimator.yaml"
And the documentation should reference "tool_collection.yaml"
And the documentation should reference "simple_graph.yaml"
And the documentation should reference "graph_workflow.yaml"
+9
View File
@@ -53,6 +53,15 @@ Feature: Actor YAML schema validation
Then the actor schema validation should succeed
And the actor config should have at least 2 tools
Scenario: Load estimator.yaml example
Given the actor YAML file "examples/actors/estimator.yaml"
When I validate the actor schema from file
Then the actor schema validation should succeed
And the actor config type should be "llm"
And the actor context_view should be "strategist"
And the actor role_hint should be "estimation"
And the actor response_format title should be "EstimationReport"
# ────────────────────────────────────────────────────────────
# Valid TOOL Actor scenarios
# ────────────────────────────────────────────────────────────
@@ -29,6 +29,13 @@ Feature: Plan Generation Pre-flight Guardrails — 7 checks before execution
When I check actor availability
Then the check should fail with message containing "estimation"
Scenario: pfg-check2-warn Estimation actor missing response_format emits warning
Given a plan preflight guardrail with all registries populated
And an estimation actor registry entry without response_format
When I run all preflight checks
Then no PreflightRejection should be raised
And the preflight warnings should contain "response_format"
# Check 3: Skill/Tool Existence
Scenario: pfg-check3 All tools and skills exist passes existence check
Given a plan preflight guardrail
+19
View File
@@ -1043,6 +1043,25 @@ def step_then_context_view_matches(context: Context, expected_view: str) -> None
assert context.actor_config.context_view.value == expected_view
@then('the actor role_hint should be "{expected_hint}"')
def step_then_role_hint_matches(context: Context, expected_hint: str) -> None:
"""Assert role hint matches."""
assert context.actor_config is not None
assert context.actor_config.role_hint is not None
assert context.actor_config.role_hint.value == expected_hint
@then('the actor response_format title should be "{expected_title}"')
def step_then_response_format_title_matches(
context: Context, expected_title: str
) -> None:
"""Assert response format title matches."""
assert context.actor_config is not None
assert context.actor_config.response_format is not None
title = context.actor_config.response_format.get("title")
assert title == expected_title
@then("the actor should have {count:d} env_vars")
def step_then_env_var_count(context: Context, count: int) -> None:
"""Assert env_vars count."""
@@ -144,6 +144,14 @@ def step_pfg_all_populated(context: Context) -> None:
context.pfg_report = None
@given("an estimation actor registry entry without response_format")
def step_pfg_estimation_actor_missing_response_format(context: Context) -> None:
context.pfg_actor_registry["estimation"] = {
"name": "local/estimator",
"context_view": "strategist",
}
@given("a plan preflight guardrail with missing action")
def step_pfg_missing_action(context: Context) -> None:
context.guardrail = PlanPreflightGuardrail()
@@ -320,3 +328,12 @@ def step_pfg_then_rejection_check(context: Context, check_name: str) -> None:
assert rejection.check.value == check_name, (
f"Expected check '{check_name}', got '{rejection.check.value}'"
)
@then('the preflight warnings should contain "{text}"')
def step_pfg_then_warnings_contain(context: Context, text: str) -> None:
report = context.pfg_report
assert report is not None, "No report recorded"
assert any(text in warning for warning in report.warnings), (
f"Expected warning containing '{text}', got {report.warnings}"
)
+12 -2
View File
@@ -14,13 +14,13 @@ ${DOCS_DIR} ${CURDIR}/../docs/reference
*** Test Cases ***
Load All Example YAML Files
[Documentation] Verify all 5 example YAML files can be loaded successfully
[Documentation] Verify all actor example YAML files can be loaded successfully
[Tags] slow
${result}= Run Process ${PYTHON} ${HELPER} count-examples ${EXAMPLES_DIR} cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} example-count:5
Should Contain ${result.stdout} example-count:8
Validate Simple LLM Example
[Documentation] Load simple_llm.yaml and verify basic LLM actor structure
@@ -40,6 +40,15 @@ Validate LLM With Tools Example
Should Contain ${result.stdout} type:llm
Should Contain ${result.stdout} has-tools
Validate Estimator Example
[Documentation] Load estimator.yaml and verify estimation actor configuration
[Tags] slow
${result}= Run Process ${PYTHON} ${HELPER} validate ${EXAMPLES_DIR}/estimator.yaml cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} actor-ok
Should Contain ${result.stdout} type:llm
Should Contain ${result.stdout} name:local/estimator
Validate Tool Collection Example
[Documentation] Load tool_collection.yaml and verify tool-only actor
[Tags] slow
@@ -109,6 +118,7 @@ Verify Documentation References All Examples
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} doc-references:simple_llm.yaml
Should Contain ${result.stdout} doc-references:llm_with_tools.yaml
Should Contain ${result.stdout} doc-references:estimator.yaml
Should Contain ${result.stdout} doc-references:tool_collection.yaml
Should Contain ${result.stdout} doc-references:simple_graph.yaml
Should Contain ${result.stdout} doc-references:graph_workflow.yaml
+1
View File
@@ -145,6 +145,7 @@ def check_docs(doc_file: str) -> int:
example_files = [
"simple_llm.yaml",
"llm_with_tools.yaml",
"estimator.yaml",
"tool_collection.yaml",
"simple_graph.yaml",
"graph_workflow.yaml",
+19 -3
View File
@@ -152,6 +152,16 @@ def _fail(msg: str) -> NoReturn:
raise SystemExit(1)
def _is_expected_provider_unavailable(output: str) -> bool:
"""Return True when failure is due to missing external LLM provider config."""
hay = output.lower()
return (
"provider openai is not configured" in hay
or "openai_api_key" in hay
or "no provider configured" in hay
)
# ---------------------------------------------------------------------------
# Subcommand: action-create
# ---------------------------------------------------------------------------
@@ -376,19 +386,25 @@ def plan_full_lifecycle() -> None:
# controlled abort with a user-facing message is expected.
r4 = run_cli("plan", "execute", plan_id, workspace=workspace)
combined4 = r4.stdout + r4.stderr
if "INTERNAL" in combined4 or "Traceback" in combined4:
if "Traceback" in combined4:
_fail(f"plan execute crashed:\n{combined4}")
if "INTERNAL" in combined4 and not _is_expected_provider_unavailable(combined4):
_fail(f"plan execute crashed:\n{combined4}")
# Step 5: Plan diff — similar: plan not in execute phase
r5 = run_cli("plan", "diff", plan_id, workspace=workspace)
combined5 = r5.stdout + r5.stderr
if "INTERNAL" in combined5 or "Traceback" in combined5:
if "Traceback" in combined5:
_fail(f"plan diff crashed:\n{combined5}")
if "INTERNAL" in combined5 and not _is_expected_provider_unavailable(combined5):
_fail(f"plan diff crashed:\n{combined5}")
# Step 6: Plan lifecycle-apply — similar: plan not ready
r6 = run_cli("plan", "lifecycle-apply", plan_id, workspace=workspace)
combined6 = r6.stdout + r6.stderr
if "INTERNAL" in combined6 or "Traceback" in combined6:
if "Traceback" in combined6:
_fail(f"plan apply crashed:\n{combined6}")
if "INTERNAL" in combined6 and not _is_expected_provider_unavailable(combined6):
_fail(f"plan apply crashed:\n{combined6}")
print("m1-plan-lifecycle-ok")
+13 -1
View File
@@ -82,6 +82,16 @@ def _fail(msg: str) -> NoReturn:
raise SystemExit(1)
def _is_expected_provider_unavailable(output: str) -> bool:
"""Return True when failure is due to missing external LLM provider config."""
hay = output.lower()
return (
"provider openai is not configured" in hay
or "openai_api_key" in hay
or "no provider configured" in hay
)
def _extract_plan_id(output: str) -> str | None:
"""Extract a ULID plan_id from plain CLI output."""
match = re.search(r"\b([0-9A-Z]{26})\b", output)
@@ -272,7 +282,9 @@ def plan_use_execute() -> None:
# correctly rejects with "not ready" (controlled abort).
r3 = run_cli("plan", "execute", plan_id, workspace=workspace)
combined = r3.stdout + r3.stderr
if "INTERNAL" in combined or "Traceback" in combined:
if "Traceback" in combined:
_fail(f"plan execute crashed:\n{combined}")
if "INTERNAL" in combined and not _is_expected_provider_unavailable(combined):
_fail(f"plan execute crashed:\n{combined}")
print("m2-plan-use-execute-ok")
+13 -1
View File
@@ -95,6 +95,16 @@ def _fail(message: str) -> NoReturn:
raise SystemExit(1)
def _is_expected_provider_unavailable(output: str) -> bool:
"""Return True when failure is due to missing external LLM provider config."""
hay = output.lower()
return (
"provider openai is not configured" in hay
or "openai_api_key" in hay
or "no provider configured" in hay
)
def _extract_plan_id(output: str) -> str | None:
"""Extract a ULID plan_id from plain CLI output."""
match = re.search(r"\b([0-9A-Z]{26})\b", output)
@@ -313,7 +323,9 @@ def plan_generates_decisions() -> None:
workspace=workspace,
)
combined = r3.stdout + r3.stderr
if "INTERNAL" in combined or "Traceback" in combined:
if "Traceback" in combined:
_fail(f"plan execute crashed:\n{combined}")
if "INTERNAL" in combined and not _is_expected_provider_unavailable(combined):
_fail(f"plan execute crashed:\n{combined}")
print("m3-plan-generates-decisions-ok")
+13 -1
View File
@@ -107,6 +107,16 @@ def _fail(msg: str) -> NoReturn:
raise SystemExit(1)
def _is_expected_provider_unavailable(output: str) -> bool:
"""Return True when failure is due to missing external LLM provider config."""
hay = output.lower()
return (
"provider openai is not configured" in hay
or "openai_api_key" in hay
or "no provider configured" in hay
)
def _extract_plan_id(output: str) -> str | None:
"""Extract a ULID plan_id from plain CLI output."""
match = re.search(r"\b([0-9A-Z]{26})\b", output)
@@ -220,7 +230,9 @@ def plan_use_execute() -> None:
# get a controlled rejection (not a crash).
r3 = run_cli("plan", "execute", plan_id, workspace=workspace)
combined = r3.stdout + r3.stderr
if "INTERNAL" in combined or "Traceback" in combined:
if "Traceback" in combined:
_fail(f"plan execute crashed:\n{combined}")
if "INTERNAL" in combined and not _is_expected_provider_unavailable(combined):
_fail(f"plan execute crashed:\n{combined}")
# Verify the output references the plan
if plan_id not in combined:
+28
View File
@@ -89,12 +89,40 @@ def _test_rollback_check() -> None:
print("rollback-check-ok")
def _test_estimation_warning() -> None:
"""Preflight emits warning for estimation actor without response_format."""
guardrail = PlanPreflightGuardrail()
report = guardrail.run_all_checks(
action_name="test-action",
action_registry={"test-action": {"name": "test-action"}},
actor_registry={
"strategy": {"name": "local/strategist"},
"execution": {"name": "local/executor"},
"estimation": {
"name": "local/estimator",
"context_view": "strategist",
},
"invariant_reconciliation": {"name": "local/reconciler"},
},
tool_names=("tool-a",),
tool_registry={"tool-a": {}},
automation_profile={"name": "auto"},
require_checkpoints=False,
validation_names=("v1",),
validation_registry={"v1": {}},
)
assert report.all_passed
assert any("response_format" in warning for warning in report.warnings)
print("estimation-warning-ok")
if __name__ == "__main__":
cmd = sys.argv[1] if len(sys.argv) > 1 else "all-pass"
dispatch: dict[str, Any] = {
"all-pass": _test_all_pass,
"action-missing": _test_action_missing,
"rollback-check": _test_rollback_check,
"estimation-warning": _test_estimation_warning,
}
fn = dispatch.get(cmd)
if fn:
+6
View File
@@ -25,3 +25,9 @@ Preflight Rollback Feasibility Check
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} rollback-check cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} rollback-check-ok
Preflight Estimation Actor Warning
[Documentation] Preflight emits warning when estimation actor lacks response_format
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} estimation-warning cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} estimation-warning-ok
+79
View File
@@ -119,6 +119,16 @@ class ContextView(StrEnum):
FULL = "full" # Complete context (use sparingly)
class RoleHint(StrEnum):
"""Optional role hint used for role-aware compatibility checks."""
STRATEGY = "strategy"
EXECUTION = "execution"
ESTIMATION = "estimation"
INVARIANT_RECONCILIATION = "invariant_reconciliation"
REVIEW = "review"
# ============================================================================
# Tool Models (for inline tool definitions in actors)
# ============================================================================
@@ -715,6 +725,10 @@ class ActorConfigSchema(BaseModel):
# LLM configuration
model: str | None = Field(default=None, description="LLM model name")
system_prompt: str | None = Field(default=None, description="System prompt")
response_format: dict[str, Any] | None = Field(
default=None,
description="Optional JSON schema for structured LLM output",
)
# Tool configuration
tools: list[str | ToolDefinition] = Field(
@@ -725,6 +739,10 @@ class ActorConfigSchema(BaseModel):
context_view: ContextView | None = Field(
default=None, description="Context filtering view"
)
role_hint: RoleHint | None = Field(
default=None,
description="Optional role hint for role-aware validation",
)
memory: MemoryConfig = Field(
default_factory=MemoryConfig, description="Memory settings"
)
@@ -877,6 +895,65 @@ class ActorConfigSchema(BaseModel):
)
def _coerce_role_hint(value: object) -> RoleHint | None:
if isinstance(value, RoleHint):
return value
if isinstance(value, str):
try:
return RoleHint(value)
except ValueError:
return None
return None
def _coerce_context_view(value: object) -> ContextView | None:
if isinstance(value, ContextView):
return value
if isinstance(value, str):
try:
return ContextView(value)
except ValueError:
return None
return None
def actor_role_warnings(config: ActorConfigSchema | dict[str, Any]) -> list[str]:
"""Return non-fatal role compatibility warnings for actor configs."""
if isinstance(config, ActorConfigSchema):
role_hint = config.role_hint
context_view = config.context_view
response_format = config.response_format
else:
role_hint = _coerce_role_hint(config.get("role_hint"))
context_view = _coerce_context_view(config.get("context_view"))
response_format = config.get("response_format")
if not isinstance(response_format, dict):
nested = config.get("config")
if isinstance(nested, dict) and isinstance(
nested.get("response_format"), dict
):
response_format = nested.get("response_format")
else:
response_format = None
if role_hint != RoleHint.ESTIMATION:
return []
warnings: list[str] = []
if not isinstance(response_format, dict) or not response_format:
warnings.append(
"Estimation actors should define 'response_format' for structured output."
)
if context_view not in (None, ContextView.STRATEGIST):
warnings.append(
"Estimation actors should use context_view 'strategist' "
"for planning context."
)
return warnings
__all__ = [
"ActorConfigSchema",
"ActorType",
@@ -888,8 +965,10 @@ __all__ = [
"NodeDefinition",
"NodeLspBinding",
"NodeType",
"RoleHint",
"RouteDefinition",
"ToolDefinition",
"ToolParameter",
"ToolSourceRef",
"actor_role_warnings",
]
@@ -66,11 +66,16 @@ class PreflightReport:
def __init__(self) -> None:
self.results: list[PreflightCheckResult] = []
self.warnings: list[str] = []
def add(self, result: PreflightCheckResult) -> None:
"""Append a check result to the report."""
self.results.append(result)
def add_warning(self, message: str) -> None:
"""Append a non-fatal warning to the report."""
self.warnings.append(message)
@property
def all_passed(self) -> bool:
"""Return ``True`` when every check passed."""
@@ -87,6 +92,8 @@ class PreflightReport:
for r in self.results:
status = "PASS" if r.passed else "FAIL"
lines.append(f" [{status}] {r.check.value}: {r.message}")
for warning in self.warnings:
lines.append(f" [WARN] actor_role_compatibility: {warning}")
return "\n".join(lines)
@@ -131,6 +138,11 @@ class PlanPreflightGuardrail:
report.add(self.check_action_schema(action_name, action_registry))
report.add(self.check_actor_availability(actor_registry))
compatibility_warning = self.check_estimation_actor_compatibility_warning(
actor_registry
)
if compatibility_warning is not None:
report.add_warning(compatibility_warning)
report.add(
self.check_skill_tool_existence(
tool_names, tool_registry, skill_names, skill_registry
@@ -232,6 +244,32 @@ class PlanPreflightGuardrail:
f"All {len(tool_names)} tools and {len(skill_names)} skills verified",
)
def check_estimation_actor_compatibility_warning(
self,
actor_registry: dict[str, object] | None,
) -> str | None:
"""Emit non-fatal warnings for estimation actor role compatibility."""
registry = actor_registry or {}
estimation_actor = registry.get("estimation")
if not isinstance(estimation_actor, dict):
return None
actor_payload = estimation_actor
response_format = actor_payload.get("response_format")
if not isinstance(response_format, dict) or not response_format:
nested = actor_payload.get("config")
if isinstance(nested, dict):
nested_response = nested.get("response_format")
if isinstance(nested_response, dict) and nested_response:
response_format = nested_response
if not isinstance(response_format, dict) or not response_format:
return (
"estimation actor is configured without response_format; "
"structured estimation output is recommended"
)
return None
def check_automation_policy(
self,
automation_profile: object | None,
+11
View File
@@ -11,6 +11,7 @@ from rich.panel import Panel
from rich.table import Table
from cleveragents.actor.config import ActorConfiguration
from cleveragents.actor.schema import actor_role_warnings
from cleveragents.application.container import get_container
from cleveragents.cli.formatting import OutputFormat, format_output
from cleveragents.core.exceptions import (
@@ -338,6 +339,12 @@ def _print_actor(
console.print(Panel(details, title=title, expand=False))
def _print_role_warnings(config_blob: dict[str, Any]) -> None:
"""Print non-fatal role compatibility warnings for actor configs."""
for warning in actor_role_warnings(config_blob):
console.print(f"[yellow]Warning:[/yellow] {warning}")
@app.command()
def add(
name: Annotated[str, typer.Argument(help="Name for the actor")],
@@ -400,6 +407,8 @@ def add(
"Actor config is marked unsafe; re-run with --unsafe to confirm."
)
_print_role_warnings(canonical_blob)
try:
if registry:
actor = registry.upsert_actor(
@@ -504,6 +513,8 @@ def update(
"Actor config is marked unsafe; re-run with --unsafe to confirm."
)
_print_role_warnings(canonical_blob)
try:
if registry:
if option_overrides is None: