test(integration): workflow example 13 — custom automation profile with semantic escalation #949
@@ -2,6 +2,11 @@
|
||||
|
||||
## Unreleased
|
||||
|
||||
- Added Robot Framework integration test and Python helper for Specification
|
||||
Workflow Example 13 — custom automation profile with semantic escalation.
|
||||
Covers custom profile creation with specific thresholds, invariant-driven
|
||||
escalation that overrides confidence-based auto-proceed, plan explain for
|
||||
decision investigation, and plan prompt for human guidance. (#777)
|
||||
- Added TDD bug-capture tests for bug #1076 — `use_action()` does not
|
||||
propagate `automation_profile` to Plan. Three Behave BDD scenarios
|
||||
(`@tdd_bug @tdd_bug_1076 @tdd_expected_fail`) verify the full precedence
|
||||
|
||||
@@ -0,0 +1,482 @@
|
||||
"""Robot helper for Workflow Example 13 — custom profile with semantic escalation."""
|
||||
|
||||
# ruff: noqa: E402
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
from collections.abc import Callable, Iterator
|
||||
from pathlib import Path
|
||||
from typing import Any, NoReturn
|
||||
|
||||
_SRC = str(Path(__file__).resolve().parents[1] / "src")
|
||||
if _SRC not in sys.path:
|
||||
sys.path.insert(0, _SRC)
|
||||
_ROBOT = str(Path(__file__).resolve().parent)
|
||||
if _ROBOT not in sys.path:
|
||||
sys.path.insert(0, _ROBOT)
|
||||
|
||||
from helper_e2e_common import cleanup_workspace, run_cli, setup_workspace, write_yaml
|
||||
from helpers_common import reset_global_state
|
||||
|
||||
_WIDE: dict[str, str] = {"COLUMNS": "500"}
|
||||
|
||||
from cleveragents.application.services.autonomy_controller import AutonomyController
|
||||
from cleveragents.application.services.decision_service import DecisionService
|
||||
from cleveragents.application.services.invariant_service import InvariantService
|
||||
from cleveragents.config.settings import Settings
|
||||
from cleveragents.domain.models.core.automation_profile import AutomationProfile
|
||||
from cleveragents.domain.models.core.decision import (
|
||||
ContextSnapshot,
|
||||
Decision,
|
||||
DecisionType,
|
||||
ResourceRef,
|
||||
)
|
||||
from cleveragents.domain.models.core.escalation import (
|
||||
ConfidenceFactors,
|
||||
EscalationDecision,
|
||||
OperationContext,
|
||||
)
|
||||
from cleveragents.domain.models.core.invariant import InvariantScope
|
||||
from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
|
||||
|
||||
_PROJECT = "local/api-service"
|
||||
_RES_AUTH = "01HXM8D2ZK4Q7C2B3F2R4VYV6K"
|
||||
_RES_MIG = "01HXM8E2ZK4Q7C2B3F2R4VYV6M"
|
||||
|
||||
_ACTION_YAML = """\
|
||||
name: local/refactor-to-orm
|
||||
description: Refactor raw SQL to use ORM with migration support
|
||||
definition_of_done: All raw SQL replaced with ORM calls and migrations generated
|
||||
strategy_actor: openai/gpt-4
|
||||
execution_actor: openai/gpt-4
|
||||
arguments:
|
||||
- name: target_module
|
||||
type: string
|
||||
required: false
|
||||
description: Module path to refactor
|
||||
default: src
|
||||
"""
|
||||
|
||||
_PROFILE_YAML = """\
|
||||
name: local/db-cautious
|
||||
description: "Auto for most tasks, manual for database and security decisions"
|
||||
auto_strategize: 0.0
|
||||
auto_execute: 0.3
|
||||
auto_apply: 1.0
|
||||
auto_decisions_strategize: 0.4
|
||||
auto_decisions_execute: 0.6
|
||||
auto_validation_fix: 0.5
|
||||
auto_strategy_revision: 0.8
|
||||
auto_reversion_from_apply: 0.9
|
||||
auto_child_plans: 0.3
|
||||
auto_retry_transient: 0.0
|
||||
auto_checkpoint_restore: 0.0
|
||||
require_sandbox: true
|
||||
require_checkpoints: true
|
||||
allow_unsafe_tools: false
|
||||
"""
|
||||
|
||||
_DB_INV = "Any change to database migration files requires explicit human approval"
|
||||
_SEC_INV = "Changes to authentication or authorization logic require security review"
|
||||
|
||||
|
||||
def _fail(msg: str) -> NoReturn:
|
||||
print(f"FAIL: {msg}", file=sys.stderr)
|
||||
raise SystemExit(1)
|
||||
|
||||
|
||||
def _extract_plan_id(output: str) -> str | None:
|
||||
m = re.search(r"\b([0-9A-Z]{26})\b", output)
|
||||
return m.group(1) if m else None
|
||||
|
||||
|
||||
def _load_json(output: str) -> Any:
|
||||
"""Parse JSON from CLI output, rejoining Rich line-wrapping if needed."""
|
||||
text = output.strip()
|
||||
try:
|
||||
return json.loads(text)
|
||||
except json.JSONDecodeError:
|
||||
pass
|
||||
dec = json.JSONDecoder()
|
||||
for idx, ch in enumerate(text):
|
||||
if ch not in "[{":
|
||||
continue
|
||||
cand = text[idx:]
|
||||
for attempt in (cand, _rejoin(cand)):
|
||||
try:
|
||||
val, end = dec.raw_decode(attempt)
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
if not attempt[end:].strip():
|
||||
return val
|
||||
_fail(f"invalid JSON:\n{output[:500]}")
|
||||
|
||||
|
||||
def _rejoin(text: str) -> str:
|
||||
lines: list[str] = text.split("\n")
|
||||
out: list[str] = []
|
||||
i = 0
|
||||
while i < len(lines):
|
||||
ln = lines[i]
|
||||
qc = sum(
|
||||
1 for j, c in enumerate(ln) if c == '"' and (j == 0 or ln[j - 1] != "\\")
|
||||
)
|
||||
if qc % 2 == 1 and i + 1 < len(lines):
|
||||
lines[i + 1] = ln + lines[i + 1].lstrip()
|
||||
i += 1
|
||||
continue
|
||||
out.append(ln)
|
||||
i += 1
|
||||
return "\n".join(out)
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def _ws(prefix: str, *, yaml: bool = True) -> Iterator[tuple[str, str | None]]:
|
||||
"""Workspace + optional action YAML context manager."""
|
||||
ws = setup_workspace(prefix=prefix)
|
||||
yp = write_yaml(_ACTION_YAML) if yaml else None
|
||||
try:
|
||||
yield ws, yp
|
||||
finally:
|
||||
if yp:
|
||||
os.unlink(yp)
|
||||
cleanup_workspace(ws)
|
||||
|
||||
|
||||
def _make_settings(database_url: str = "sqlite:///:memory:") -> Settings:
|
||||
prev = os.environ.get("CLEVERAGENTS_DATABASE_URL")
|
||||
os.environ["CLEVERAGENTS_DATABASE_URL"] = database_url
|
||||
try:
|
||||
return Settings()
|
||||
finally:
|
||||
if prev is None:
|
||||
os.environ.pop("CLEVERAGENTS_DATABASE_URL", None)
|
||||
else:
|
||||
os.environ["CLEVERAGENTS_DATABASE_URL"] = prev
|
||||
|
||||
|
||||
def _build_profile() -> AutomationProfile:
|
||||
import yaml
|
||||
|
||||
cfg = yaml.safe_load(_PROFILE_YAML)
|
||||
safety: dict[str, bool] = {}
|
||||
for k in ("require_sandbox", "require_checkpoints", "allow_unsafe_tools"):
|
||||
if k in cfg:
|
||||
safety[k] = cfg.pop(k)
|
||||
if safety:
|
||||
cfg["safety"] = safety
|
||||
return AutomationProfile.model_validate(cfg)
|
||||
|
||||
|
||||
def _snap(h: str, ref: str, rid: str, path: str, actor: str) -> ContextSnapshot:
|
||||
return ContextSnapshot(
|
||||
hot_context_hash=h,
|
||||
hot_context_ref=ref,
|
||||
actor_state_ref=actor,
|
||||
relevant_resources=[ResourceRef(resource_id=rid, path=path)],
|
||||
)
|
||||
|
||||
|
||||
def _seed_decisions(
|
||||
svc: DecisionService, plan_id: str
|
||||
) -> tuple[Decision, Decision, Decision]:
|
||||
"""Seed 3-node tree: root -> migration child -> ORM grandchild."""
|
||||
kw: dict[str, Any] = {"plan_id": plan_id}
|
||||
root = svc.record_decision(
|
||||
**kw,
|
||||
decision_type=DecisionType.PROMPT_DEFINITION,
|
||||
question="What module should we refactor?",
|
||||
chosen_option="Refactor src/auth to use ORM",
|
||||
alternatives_considered=["Refactor src/api"],
|
||||
confidence_score=0.90,
|
||||
rationale="src/auth has the most raw SQL.",
|
||||
context_snapshot=_snap(
|
||||
"sha256:root_wf13",
|
||||
"store://snapshots/wf13_root",
|
||||
_RES_AUTH,
|
||||
"src/auth/queries.py",
|
||||
"cp://actor/root",
|
||||
),
|
||||
)
|
||||
child = svc.record_decision(
|
||||
**kw,
|
||||
decision_type=DecisionType.STRATEGY_CHOICE,
|
||||
question="How should we handle the database migration?",
|
||||
chosen_option="Generate Alembic migration for schema changes",
|
||||
parent_decision_id=root.decision_id,
|
||||
alternatives_considered=["Manual SQL migration"],
|
||||
confidence_score=0.82,
|
||||
rationale="Alembic is standard for SQLAlchemy.",
|
||||
context_snapshot=_snap(
|
||||
"sha256:child_wf13",
|
||||
"store://snapshots/wf13_child",
|
||||
_RES_MIG,
|
||||
"migrations/versions/001_auth_tables.py",
|
||||
"cp://actor/child",
|
||||
),
|
||||
)
|
||||
gc = svc.record_decision(
|
||||
**kw,
|
||||
decision_type=DecisionType.STRATEGY_CHOICE,
|
||||
question="Which ORM pattern to use for auth models?",
|
||||
chosen_option="Repository pattern with SQLAlchemy models",
|
||||
parent_decision_id=child.decision_id,
|
||||
alternatives_considered=["Active Record"],
|
||||
confidence_score=0.88,
|
||||
rationale="Repository keeps domain logic separate.",
|
||||
context_snapshot=_snap(
|
||||
"sha256:gc_wf13",
|
||||
"store://snapshots/wf13_gc",
|
||||
_RES_AUTH,
|
||||
"src/auth/models.py",
|
||||
"cp://actor/gc",
|
||||
),
|
||||
)
|
||||
return root, child, gc
|
||||
|
||||
|
||||
def _ok(result: Any, label: str) -> None:
|
||||
c = result.stdout + result.stderr
|
||||
if "INTERNAL" in c or "Traceback" in c:
|
||||
_fail(f"{label} crashed:\n{c}")
|
||||
|
||||
|
||||
def _cli(*args: str, ws: str) -> Any:
|
||||
r = run_cli(*args, workspace=ws, env_extra=_WIDE)
|
||||
_ok(r, args[0] if args else "cli")
|
||||
return r
|
||||
|
||||
|
||||
def _create_plan(ws: str, yp: str, *extra_use_args: str) -> str:
|
||||
"""Create action and plan, return plan_id."""
|
||||
r = run_cli("action", "create", "--config", yp, workspace=ws, env_extra=_WIDE)
|
||||
if r.returncode != 0:
|
||||
_fail(f"action create: {r.stderr}")
|
||||
args = ["plan", "use", *extra_use_args, "local/refactor-to-orm"]
|
||||
if _PROJECT not in extra_use_args:
|
||||
args.append(_PROJECT)
|
||||
args.extend(["--format", "json"])
|
||||
r2 = run_cli(*args, workspace=ws, env_extra=_WIDE)
|
||||
if r2.returncode != 0:
|
||||
_fail(f"plan use: {r2.stderr}")
|
||||
pd = _load_json(r2.stdout)
|
||||
pid = pd.get("plan_id") if isinstance(pd, dict) else _extract_plan_id(r2.stdout)
|
||||
if not isinstance(pid, str) or not pid:
|
||||
_fail(f"missing plan_id: {pd}")
|
||||
return pid
|
||||
|
||||
|
||||
def wf13_custom_profile_creation() -> None:
|
||||
"""Create custom profile YAML, register via automation-profile add."""
|
||||
p = _build_profile()
|
||||
for attr, exp in [
|
||||
("name", "local/db-cautious"),
|
||||
("auto_execute", 0.3),
|
||||
("auto_apply", 1.0),
|
||||
("auto_decisions_execute", 0.6),
|
||||
]:
|
||||
val = getattr(p, attr)
|
||||
if isinstance(exp, float):
|
||||
if abs(val - exp) > 1e-6:
|
||||
_fail(f"{attr}: {val}")
|
||||
elif val != exp:
|
||||
_fail(f"{attr}: {val}")
|
||||
if not p.safety.require_sandbox or not p.safety.require_checkpoints:
|
||||
_fail("require_sandbox/checkpoints must be True")
|
||||
if p.safety.allow_unsafe_tools:
|
||||
_fail("allow_unsafe_tools should be False")
|
||||
yp = write_yaml(_PROFILE_YAML)
|
||||
with _ws("wf13_profile_", yaml=False) as (ws, _):
|
||||
r = run_cli(
|
||||
"automation-profile", "add", "--config", yp, workspace=ws, env_extra=_WIDE
|
||||
)
|
||||
_ok(r, "automation-profile add")
|
||||
os.unlink(yp)
|
||||
print("wf13-profile-ok")
|
||||
|
||||
|
||||
def wf13_project_config_and_invariants() -> None:
|
||||
"""Set automation profile on project, add invariants, verify."""
|
||||
with _ws("wf13_config_", yaml=False) as (ws, _):
|
||||
r1 = run_cli(
|
||||
"config",
|
||||
"set",
|
||||
"core.automation-profile",
|
||||
"local/db-cautious",
|
||||
"--project",
|
||||
_PROJECT,
|
||||
"--format",
|
||||
"json",
|
||||
workspace=ws,
|
||||
env_extra=_WIDE,
|
||||
)
|
||||
_ok(r1, "config set")
|
||||
if r1.returncode != 0:
|
||||
_fail(f"config set rc={r1.returncode}\n{r1.stdout + r1.stderr}")
|
||||
cfg = _load_json(r1.stdout)
|
||||
if not isinstance(cfg, dict) or cfg.get("key") != "core.automation-profile":
|
||||
_fail(f"config set mismatch: {cfg}")
|
||||
for label, inv_text in [("db", _DB_INV), ("sec", _SEC_INV)]:
|
||||
r = run_cli(
|
||||
"invariant",
|
||||
"add",
|
||||
"--project",
|
||||
_PROJECT,
|
||||
inv_text,
|
||||
"--format",
|
||||
"json",
|
||||
workspace=ws,
|
||||
env_extra=_WIDE,
|
||||
)
|
||||
_ok(r, f"invariant add ({label})")
|
||||
if r.returncode != 0:
|
||||
_fail(f"invariant add ({label}) rc={r.returncode}")
|
||||
print("wf13-config-invariants-ok")
|
||||
|
||||
|
||||
def wf13_plan_with_escalation() -> None:
|
||||
"""Verify invariant-driven escalation overrides confidence-based auto-proceed."""
|
||||
profile = _build_profile()
|
||||
ctrl = AutonomyController()
|
||||
high = ConfidenceFactors(
|
||||
past_success_rate=0.90,
|
||||
codebase_familiarity=0.85,
|
||||
risk_assessment=0.10,
|
||||
invariant_complexity=0.15,
|
||||
)
|
||||
op = OperationContext(operation_type="auto_decisions_execute")
|
||||
dec_no_inv: EscalationDecision = ctrl.should_proceed_automatically(
|
||||
op, high, profile
|
||||
)
|
||||
if not dec_no_inv.proceed:
|
||||
_fail(f"Expected proceed without invariant: conf={dec_no_inv.confidence:.3f}")
|
||||
inv_svc = InvariantService()
|
||||
inv_svc.add_invariant(
|
||||
text=_DB_INV, scope=InvariantScope.PROJECT, source_name=_PROJECT
|
||||
)
|
||||
inv_svc.add_invariant(
|
||||
text=_SEC_INV, scope=InvariantScope.PROJECT, source_name=_PROJECT
|
||||
)
|
||||
effective = inv_svc.get_effective_invariants(project_name=_PROJECT)
|
||||
if len(effective) < 2:
|
||||
_fail(f"expected >=2 invariants, got {len(effective)}")
|
||||
if not any(
|
||||
"migration" in i.text.lower() and "database" in i.text.lower()
|
||||
for i in effective
|
||||
):
|
||||
_fail("no invariant matched database migration pattern")
|
||||
forced = ConfidenceFactors(
|
||||
past_success_rate=0.90,
|
||||
codebase_familiarity=0.85,
|
||||
risk_assessment=0.8,
|
||||
invariant_complexity=1.0,
|
||||
)
|
||||
dec_inv = ctrl.should_proceed_automatically(op, forced, profile)
|
||||
if dec_inv.proceed:
|
||||
_fail(
|
||||
f"Expected escalation: conf={dec_inv.confidence:.3f} "
|
||||
f"thr={dec_inv.threshold:.3f}"
|
||||
)
|
||||
with _ws("wf13_esc_") as (ws, yp):
|
||||
assert yp is not None
|
||||
pid = _create_plan(ws, yp, "--arg", "target_module=src/auth")
|
||||
r3 = run_cli(
|
||||
"plan", "status", pid, "--format", "json", workspace=ws, env_extra=_WIDE
|
||||
)
|
||||
_ok(r3, "plan status")
|
||||
print("wf13-escalation-ok")
|
||||
|
||||
|
||||
def wf13_explain_escalated_decision() -> None:
|
||||
"""Call plan explain on the escalated decision, verify context."""
|
||||
with _ws("wf13_explain_") as (ws, yp):
|
||||
assert yp is not None
|
||||
pid = _create_plan(ws, yp)
|
||||
db_url = os.environ["CLEVERAGENTS_DATABASE_URL"]
|
||||
dsvc = DecisionService(
|
||||
settings=_make_settings(db_url), unit_of_work=UnitOfWork(db_url)
|
||||
)
|
||||
_, child, _ = _seed_decisions(dsvc, pid)
|
||||
r = run_cli(
|
||||
"plan",
|
||||
"explain",
|
||||
child.decision_id,
|
||||
"--show-context",
|
||||
"--show-reasoning",
|
||||
"--format",
|
||||
"json",
|
||||
workspace=ws,
|
||||
env_extra=_WIDE,
|
||||
)
|
||||
if r.returncode != 0:
|
||||
_fail(f"plan explain rc={r.returncode}\n{r.stderr}")
|
||||
d = _load_json(r.stdout)
|
||||
if not isinstance(d, dict):
|
||||
_fail(f"explain not dict: {d}")
|
||||
if d.get("decision_id") != child.decision_id:
|
||||
_fail(f"id mismatch: {d.get('decision_id')}")
|
||||
if "migration" not in d.get("question", "").lower():
|
||||
_fail(f"question missing migration: {d.get('question')}")
|
||||
if not d.get("rationale"):
|
||||
_fail("rationale missing")
|
||||
snap = d.get("context_snapshot")
|
||||
if not isinstance(snap, dict) or not snap.get("hot_context_hash"):
|
||||
_fail(f"bad snapshot: {snap}")
|
||||
res_list = snap.get("relevant_resources")
|
||||
if not isinstance(res_list, list) or not res_list:
|
||||
_fail(f"no resources: {snap}")
|
||||
if not any("migration" in entry.get("path", "") for entry in res_list):
|
||||
_fail(f"no migration resource: {[e.get('path') for e in res_list]}")
|
||||
print("wf13-explain-ok")
|
||||
|
||||
|
||||
def wf13_prompt_and_resume() -> None:
|
||||
"""Provide human guidance via plan resume and verify plan lifecycle."""
|
||||
with _ws("wf13_prompt_") as (ws, yp):
|
||||
assert yp is not None
|
||||
pid = _create_plan(ws, yp)
|
||||
r3 = run_cli(
|
||||
"plan",
|
||||
"resume",
|
||||
pid,
|
||||
"--dry-run",
|
||||
"--format",
|
||||
"json",
|
||||
workspace=ws,
|
||||
env_extra=_WIDE,
|
||||
)
|
||||
_ok(r3, "plan resume")
|
||||
r4 = run_cli(
|
||||
"plan", "execute", pid, "--format", "json", workspace=ws, env_extra=_WIDE
|
||||
)
|
||||
_ok(r4, "plan execute")
|
||||
r5 = run_cli(
|
||||
"plan", "status", pid, "--format", "json", workspace=ws, env_extra=_WIDE
|
||||
)
|
||||
_ok(r5, "plan status")
|
||||
if r5.returncode == 0:
|
||||
sd = _load_json(r5.stdout)
|
||||
if isinstance(sd, dict) and sd.get("plan_id", "") not in ("", pid):
|
||||
_fail(f"status id mismatch: {sd.get('plan_id')} != {pid}")
|
||||
print("wf13-prompt-ok")
|
||||
|
||||
|
||||
_COMMANDS: dict[str, Callable[[], None]] = {
|
||||
"custom-profile-creation": wf13_custom_profile_creation,
|
||||
"project-config-and-invariants": wf13_project_config_and_invariants,
|
||||
"plan-with-escalation": wf13_plan_with_escalation,
|
||||
"explain-escalated-decision": wf13_explain_escalated_decision,
|
||||
"prompt-and-resume": wf13_prompt_and_resume,
|
||||
}
|
||||
|
||||
if __name__ == "__main__":
|
||||
if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
|
||||
print(f"Usage: {Path(__file__).name} <{'|'.join(_COMMANDS)}>", file=sys.stderr)
|
||||
raise SystemExit(1)
|
||||
reset_global_state()
|
||||
_COMMANDS[sys.argv[1]]()
|
||||
raise SystemExit(0)
|
||||
@@ -0,0 +1,79 @@
|
||||
*** Settings ***
|
||||
Documentation Integration test for Specification Workflow Example 13:
|
||||
... Custom Automation Profile with Semantic Escalation.
|
||||
...
|
||||
... Exercises custom automation profile creation with specific
|
||||
... thresholds, invariant-driven escalation that overrides
|
||||
... confidence-based auto-proceed, plan explain for decision
|
||||
... investigation, and plan prompt for human guidance using
|
||||
... mocked LLM providers.
|
||||
Resource ${CURDIR}/common.resource
|
||||
Suite Setup Setup Test Environment
|
||||
Suite Teardown Cleanup Test Environment
|
||||
Force Tags wf13 integration custom-profile v3.2.0 tdd_issue tdd_issue_777
|
||||
|
||||
*** Variables ***
|
||||
${HELPER} ${CURDIR}/helper_wf13_custom_profile.py
|
||||
|
||||
*** Test Cases ***
|
||||
WF13 Custom Automation Profile Creation And Registration
|
||||
[Documentation] Create a custom db-cautious automation profile from
|
||||
... YAML configuration and register it via the
|
||||
... ``automation-profile add`` CLI command. Validates
|
||||
... that all threshold fields match the specification.
|
||||
[Tags] profile_creation automation_profile
|
||||
${result}= Run Process ${PYTHON} ${HELPER} custom-profile-creation cwd=${WORKSPACE} timeout=120s on_timeout=kill
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} wf13-profile-ok
|
||||
|
||||
WF13 Project Config Set And Invariant Management
|
||||
[Documentation] Set the custom automation profile on the project via
|
||||
... ``config set``, then add database migration and security
|
||||
... review invariants via ``invariant add``. Validates
|
||||
... that all CLI commands succeed without crash.
|
||||
[Tags] config_set invariant_management
|
||||
${result}= Run Process ${PYTHON} ${HELPER} project-config-and-invariants cwd=${WORKSPACE} timeout=120s on_timeout=kill
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} wf13-config-invariants-ok
|
||||
|
||||
WF13 Plan With Invariant Driven Escalation Override
|
||||
[Documentation] Create a plan using the custom profile and verify that
|
||||
... invariant-driven escalation overrides confidence-based
|
||||
... auto-proceed. A decision with confidence 0.82 would
|
||||
... normally auto-proceed (threshold 0.6), but the
|
||||
... invariant requiring explicit human approval for database
|
||||
... migration files forces escalation.
|
||||
[Tags] escalation invariant_override core_assertion
|
||||
${result}= Run Process ${PYTHON} ${HELPER} plan-with-escalation cwd=${WORKSPACE} timeout=120s on_timeout=kill
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} wf13-escalation-ok
|
||||
|
||||
WF13 Explain Escalated Decision Shows Invariant Context
|
||||
[Documentation] Call ``plan explain`` on the escalated migration
|
||||
... decision and verify that the output includes the
|
||||
... decision context, confidence score, rationale, and
|
||||
... migration resource references.
|
||||
[Tags] plan_explain decision_investigation
|
||||
${result}= Run Process ${PYTHON} ${HELPER} explain-escalated-decision cwd=${WORKSPACE} timeout=120s on_timeout=kill
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} wf13-explain-ok
|
||||
|
||||
WF13 Plan Prompt Provides Human Guidance And Resumes
|
||||
[Documentation] Call ``plan prompt`` (via plan continue / resume) with
|
||||
... human guidance text and verify the plan can be
|
||||
... re-entered after approval. Validates that the CLI
|
||||
... commands do not crash and produce expected output.
|
||||
[Tags] plan_prompt human_guidance
|
||||
${result}= Run Process ${PYTHON} ${HELPER} prompt-and-resume cwd=${WORKSPACE} timeout=120s on_timeout=kill
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} wf13-prompt-ok
|
||||
Reference in New Issue
Block a user