cleveragents/cleveragents-core
4
0
Fork 3
Code Issues 6.0k Pull Requests 115 Actions Packages Projects Releases Wiki Activity

fix(db): align v3_plans schema with specification DDL #1074

Merged
CoreRasurae merged 1 commits from fix/plans-table-schema-alignment into master 2026-03-30 23:01:20 +00:00
Conversation 16 Commits 1 Files Changed 22
+1569 -39
22 changed files with 1569 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+40
View File
@@ -436,6 +436,46 @@
in `execute_plan()` for consistency with phase-state validator.
Updated `PlanResumeService` docstring to reflect ERRORED
terminality. (#918)
- Aligned `v3_plans` table schema with specification DDL: added
`effective_profile_snapshot` column (TEXT NOT NULL, validated as JSON),
made `root_plan_id` NOT NULL with self-referencing for root plans and
explicit `RESTRICT` FK policy, made `automation_profile` NOT NULL with
default `"balanced"`, and documented the intentional `phase` default
deviation (`"action"` vs spec `"strategize"`). Migration backfill
correctly resolves root ancestors for child plans at arbitrary hierarchy
depth via level-by-level propagation with safety bound. Hardened
`automation_profile` deserialization to catch `RecursionError`.
Hardened `effective_profile_snapshot` deserialization in `to_domain()`
to gracefully fall back to `'{}'` on corrupted JSON, preventing a
single corrupted row from crashing plan reads. Added `RecursionError`
to `effective_profile_snapshot` Pydantic validator for consistency
with `automation_profile` deserialization. Validator error message
now uses length-only to avoid potential information disclosure.
Documented intentional column naming conventions vs spec DDL.
Migration orphan-row fallback now logs affected row count.
Documented FK ondelete policy drift between ORM and migrated schemas.
Documented `automation_profile` dual-format storage semantic.
Added `TypeError` to `effective_profile_snapshot` deserialization
exception list in `to_domain()` for consistency with the Pydantic
validator. Extracted default automation profile name to a
module-level constant (`DEFAULT_AUTOMATION_PROFILE`) to reduce
sentinel duplication across `models.py` and `repositories.py`.
Migration cycle-detection now logs affected `plan_id` values
(truncated to first 50) before the orphan fallback runs.
Migration SQL statements
uniformly use `sa.text()` for consistency. Centralised
automation-profile serialisation into
`LifecyclePlanModel._serialize_automation_profile()` to
eliminate duplication between `from_domain()` and
`LifecyclePlanRepository.update()`.
Moved `root_plan_id` self-reference resolution from
`from_domain()` into a `PlanIdentity` `model_validator` so
the domain model is consistent with the DB ``NOT NULL``
constraint before and after persistence.
Used explicit ``is not None`` check in
`_serialize_automation_profile()` for consistency with the
explicit-None-check convention used elsewhere in this commit.
(#921)
- Fixed `shell=True` subprocess usage in `cli_coverage_steps.py` by replacing
with `shlex.split()` and `shell=False` for defense-in-depth command injection
prevention, consistent with the existing pattern in
alembic/versions/m8_001_align_plans_schema.py
+152
View File
@@ -0,0 +1,152 @@
"""Align v3_plans table schema with specification DDL.
- Add effective_profile_snapshot column (TEXT NOT NULL DEFAULT '{}').
- Make root_plan_id NOT NULL (backfill with plan_id first).
- Make automation_profile NOT NULL (backfill with 'balanced' first).
Note on downgrade: the backfill is not reversible. After downgrade,
root plans that previously had ``root_plan_id IS NULL`` will retain
the self-referencing value (``root_plan_id = plan_id``).
Revision ID: m8_001_align_plans_schema
Revises: m4_003_plan_env_columns
Create Date: 2026-03-19 00:00:00
"""
import logging
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
_logger = logging.getLogger(__name__)
# revision identifiers, used by Alembic.
revision: str = "m8_001_align_plans_schema"
down_revision: str | Sequence[str] | None = "m4_003_plan_env_columns"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
"""Align v3_plans with spec DDL: new column + nullability fixes."""
# 1. Add effective_profile_snapshot column
op.add_column(
"v3_plans",
sa.Column(
"effective_profile_snapshot",
sa.Text(),
nullable=False,
server_default="{}",
),
)
# 2. Backfill root_plan_id: root plans (no parent) self-reference;
# child plans inherit root from their parent's root_plan_id.
# The child backfill propagates level-by-level: each iteration
# only updates children whose parent already has root_plan_id
# resolved. This avoids the snapshot-semantics pitfall where a
# single correlated UPDATE would give grandchildren their
# parent's plan_id instead of the root ancestor's ID.
op.execute(
sa.text(
"UPDATE v3_plans SET root_plan_id = plan_id "
"WHERE root_plan_id IS NULL AND parent_plan_id IS NULL"
)
)
bind = op.get_bind()
_MAX_DEPTH = 100
for _depth in range(_MAX_DEPTH):
result = bind.execute(
sa.text(
"UPDATE v3_plans SET root_plan_id = ("
" SELECT p.root_plan_id"
" FROM v3_plans p WHERE p.plan_id = v3_plans.parent_plan_id"
") WHERE root_plan_id IS NULL"
" AND parent_plan_id IS NOT NULL"
" AND parent_plan_id IN ("
" SELECT p2.plan_id FROM v3_plans p2"
" WHERE p2.root_plan_id IS NOT NULL"
" )"
)
)
if result.rowcount == 0:
break
else:
# Log affected plan IDs before the fallback overwrites them.
cycle_rows = bind.execute(
sa.text("SELECT plan_id FROM v3_plans WHERE root_plan_id IS NULL")
).fetchall()
cycle_ids = [r[0] for r in cycle_rows]
_MAX_LOG_IDS = 50
display_ids = cycle_ids[:_MAX_LOG_IDS]
_logger.error(
"root_plan_id backfill did not converge after %d iterations "
"— possible cycle in parent_plan_id; affected count=%d, "
"plan_ids (first %d)=%s",
_MAX_DEPTH,
len(cycle_ids),
_MAX_LOG_IDS,
display_ids,
)
# Fallback: any remaining NULLs (orphaned children) self-reference.
fallback_result = bind.execute(
sa.text("UPDATE v3_plans SET root_plan_id = plan_id WHERE root_plan_id IS NULL")
)
if fallback_result.rowcount > 0:
_logger.warning(
"root_plan_id fallback: %d orphaned row(s) self-referenced "
"(parent_plan_id points to a missing plan)",
fallback_result.rowcount,
)
# 3. Backfill automation_profile with 'balanced' where NULL or empty.
op.execute(
sa.text(
"UPDATE v3_plans SET automation_profile = 'balanced' "
"WHERE automation_profile IS NULL OR automation_profile = ''"
)
)
# 4. Apply NOT NULL constraints in a single batch to avoid
# two full table copies in SQLite batch mode.
#
# NOTE: The ORM model specifies ondelete="RESTRICT" for the
# root_plan_id FK, but SQLite batch_alter_table recreates the
# table from reflected metadata and does not update FK policies
# unless explicitly told to. Fresh databases created via
# Base.metadata.create_all() get the correct RESTRICT policy
# from the ORM model. Migrated databases retain the previous
# FK policy. This is acceptable because SQLite disables FK
# enforcement by default (PRAGMA foreign_keys = OFF) and the
# NOT NULL constraint on root_plan_id independently prevents
# the SET NULL action from succeeding.
with op.batch_alter_table("v3_plans") as batch_op:
batch_op.alter_column(
"root_plan_id", existing_type=sa.String(26), nullable=False
)
batch_op.alter_column(
"automation_profile",
existing_type=sa.Text(),
nullable=False,
server_default="balanced",
)
def downgrade() -> None:
"""Revert v3_plans schema changes."""
# Reverse order: make columns nullable again, drop new column.
# Single batch to minimise table copies.
with op.batch_alter_table("v3_plans") as batch_op:
batch_op.alter_column(
"automation_profile",
existing_type=sa.Text(),
nullable=True,
server_default=None,
)
batch_op.alter_column(
"root_plan_id", existing_type=sa.String(26), nullable=True
)
op.drop_column("v3_plans", "effective_profile_snapshot")
alembic/versions/m8_002_merge_profile_rename_and_corrections.py
+7 -5
View File
@@ -1,11 +1,12 @@
"""Merge profile-rename and correction-attempts heads.
"""Merge profile-rename, correction-attempts, and plans-schema heads.
Both m5_001_rename_profile_fields and m8_001_correction_attempts branched
from m4_003_plan_env_columns, creating two Alembic heads. This no-op
merge migration resolves them into a single head.
m5_001_rename_profile_fields, m8_001_correction_attempts, and
m8_001_align_plans_schema all branched from m4_003_plan_env_columns,
creating three Alembic heads. This no-op merge migration resolves
them into a single head.
Revision ID: m8_002_merge_profile_rename_and_corrections
Revises: m5_001_rename_profile_fields, m8_001_correction_attempts
Revises: m5_001_rename_profile_fields, m8_001_correction_attempts, m8_001_align_plans_schema
Create Date: 2026-03-29 00:00:00
"""
@@ -17,6 +18,7 @@ revision: str = "m8_002_merge_profile_rename_and_corrections"
down_revision: str | Sequence[str] | None = (
"m5_001_rename_profile_fields",
"m8_001_correction_attempts",
"m8_001_align_plans_schema",
)
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
benchmarks/plan_phase_migration_bench.py
+12 -3
View File
@@ -86,7 +86,9 @@ class PlanInsertActionPhase:
now = _now_iso()
for i in range(100):
plan = LifecyclePlanModel()
plan.plan_id = _make_ulid(base + i)
pid = _make_ulid(base + i)
plan.plan_id = pid
plan.root_plan_id = pid
plan.action_name = "bench/phase-action"
plan.namespaced_name = f"bench/plan-{it}-{i}"
plan.namespace = "bench"
@@ -94,6 +96,7 @@ class PlanInsertActionPhase:
plan.processing_state = "queued"
plan.description = f"Benchmark plan {it}-{i}"
plan.tags_json = "[]"
plan.effective_profile_snapshot = "{}"
plan.created_at = now
plan.updated_at = now
session.add(plan)
@@ -136,7 +139,9 @@ class PlanInsertApplyTerminalStates:
now = _now_iso()
for i in range(50):
plan = LifecyclePlanModel()
plan.plan_id = _make_ulid(base + i)
pid = _make_ulid(base + i)
plan.plan_id = pid
plan.root_plan_id = pid
plan.action_name = "bench/terminal-action"
plan.namespaced_name = f"bench/applied-{it}-{i}"
plan.namespace = "bench"
@@ -144,6 +149,7 @@ class PlanInsertApplyTerminalStates:
plan.processing_state = "applied"
plan.description = f"Applied plan {it}-{i}"
plan.tags_json = "[]"
plan.effective_profile_snapshot = "{}"
plan.created_at = now
plan.updated_at = now
session.add(plan)
@@ -157,7 +163,9 @@ class PlanInsertApplyTerminalStates:
now = _now_iso()
for i in range(50):
plan = LifecyclePlanModel()
plan.plan_id = _make_ulid(base + i)
pid = _make_ulid(base + i)
plan.plan_id = pid
plan.root_plan_id = pid
plan.action_name = "bench/terminal-action"
plan.namespaced_name = f"bench/constrained-{it}-{i}"
plan.namespace = "bench"
@@ -165,6 +173,7 @@ class PlanInsertApplyTerminalStates:
plan.processing_state = "constrained"
plan.description = f"Constrained plan {it}-{i}"
plan.tags_json = "[]"
plan.effective_profile_snapshot = "{}"
plan.created_at = now
plan.updated_at = now
session.add(plan)
features/plans_table_schema_alignment.feature
+139
View File
@@ -0,0 +1,139 @@
@feature921
Feature: v3_plans table schema alignment with spec DDL
As a database administrator
I want the v3_plans table schema to match the specification DDL
So that nullability, defaults, and columns are correct
Background:
Given a schema-aligned fresh in-memory database
And a schema-aligned prerequisite action "local/schema-test" exists
@feature921
Scenario: Plan creation includes effective_profile_snapshot
Given a schema-aligned plan with ID "01HV000000000000000000SA01"
And the plan effective_profile_snapshot is '{"profile":"balanced","thresholds":{}}'
When I persist the schema-aligned plan
Then the retrieved plan effective_profile_snapshot should be '{"profile":"balanced","thresholds":{}}'
@feature921
Scenario: Plan creation with default effective_profile_snapshot
Given a schema-aligned plan with ID "01HV000000000000000000SA02"
When I persist the schema-aligned plan
Then the retrieved plan effective_profile_snapshot should be '{}'
@feature921
Scenario: Root plan has self-referencing root_plan_id
Given a schema-aligned root plan with ID "01HV000000000000000000SA03"
When I persist the schema-aligned plan
Then the plan root_plan_id should equal the plan_id "01HV000000000000000000SA03"
@feature921
Scenario: Child plan has parent root_plan_id
Given a schema-aligned root plan with ID "01HV000000000000000000SA04"
And a schema-aligned child plan with ID "01HV000000000000000000SA05" parent "01HV000000000000000000SA04" root "01HV000000000000000000SA04"
When I persist both schema-aligned plans
Then the child plan root_plan_id should be "01HV000000000000000000SA04"
And the child plan parent_plan_id should be "01HV000000000000000000SA04"
@feature921
Scenario: automation_profile defaults to balanced for plans without profile
Given a schema-aligned plan with ID "01HV000000000000000000SA06" and no automation profile
When I persist the schema-aligned plan
Then the persisted automation_profile column should be "balanced"
@feature921
Scenario: Bare-string automation_profile deserializes gracefully
Given a schema-aligned plan with ID "01HV000000000000000000SA07"
And the plan automation_profile column is set to bare string "full-auto"
When I load the plan from the database
Then the plan automation_profile ref should be None
@feature921
Scenario: Invalid provenance in automation_profile JSON deserializes gracefully
Given a schema-aligned plan with ID "01HV000000000000000000SA08"
And the plan automation_profile column is set to '{"profile_name":"test","provenance":"bogus"}'
When I load the plan from the database
Then the plan automation_profile ref should be None
@feature921
Scenario: Grandchild plan carries root ancestor plan_id
Given a schema-aligned root plan with ID "01HV000000000000000000SA10"
And a schema-aligned child plan with ID "01HV000000000000000000SA11" parent "01HV000000000000000000SA10" root "01HV000000000000000000SA10"
And a schema-aligned grandchild plan with ID "01HV000000000000000000SA12" parent "01HV000000000000000000SA11" root "01HV000000000000000000SA10"
When I persist root, child, and grandchild schema-aligned plans
Then the grandchild plan root_plan_id should be "01HV000000000000000000SA10"
And the grandchild plan parent_plan_id should be "01HV000000000000000000SA11"
@feature921
Scenario: automation_profile None roundtrips as None through domain model
Given a schema-aligned plan with ID "01HV000000000000000000SA13" and no automation profile
When I persist the schema-aligned plan
Then the persisted automation_profile column should be "balanced"
And the plan automation_profile ref should be None
@feature921
Scenario: Truncated JSON in automation_profile deserializes gracefully
Given a schema-aligned plan with ID "01HV000000000000000000SA14"
And the plan automation_profile column is set to '{"profile_name":"test"'
When I load the plan from the database
Then the plan automation_profile ref should be None
@feature921
Scenario: Updating effective_profile_snapshot persists correctly
Given a schema-aligned plan with ID "01HV000000000000000000SA15"
And the plan effective_profile_snapshot is '{"profile":"balanced","thresholds":{}}'
When I persist the schema-aligned plan
And I update the plan effective_profile_snapshot to '{"profile":"full-auto","thresholds":{"strategy":0.5}}'
Then the retrieved plan effective_profile_snapshot should be '{"profile":"full-auto","thresholds":{"strategy":0.5}}'
@feature921
Scenario: AutomationProfileRef round-trips through persistence
Given a schema-aligned plan "01HV000000000000000000SA16" with profile "supervised" and provenance "plan"
When I persist the schema-aligned plan
Then the plan automation_profile name should be "supervised"
And the plan automation_profile provenance should be "plan"
@feature921
Scenario: NULL root_plan_id is rejected by the database
Given a schema-aligned plan with ID "01HV000000000000000000SA09" forced NULL root_plan_id
When I attempt to persist the plan with NULL root_plan_id
Then the database should reject the insert with an integrity error
@feature921
Scenario: NULL effective_profile_snapshot is rejected by the database
Given a schema-aligned plan with ID "01HV000000000000000000SA17" forced NULL effective_profile_snapshot
When I attempt to persist the plan with NULL effective_profile_snapshot
Then the database should reject the snapshot insert with an integrity error
@feature921
Scenario: PlanIdentity resolves None root_plan_id to plan_id at construction
Given a schema-aligned plan with ID "01HV000000000000000000SA18" and None root_plan_id
Then the domain plan root_plan_id should already equal "01HV000000000000000000SA18"
When I convert the plan via from_domain
Then the ORM model root_plan_id should equal "01HV000000000000000000SA18"
@feature921
Scenario: Valid JSON missing profile_name key deserializes gracefully
Given a schema-aligned plan with ID "01HV000000000000000000SA19"
And the plan automation_profile column is set to '{"provenance":"plan"}'
When I load the plan from the database
Then the plan automation_profile ref should be None
@feature921
Scenario: Corrupted effective_profile_snapshot in DB falls back to empty JSON
Given a schema-aligned plan with ID "01HV000000000000000000SA20"
And the plan effective_profile_snapshot is '{"profile":"balanced"}'
When I persist the schema-aligned plan
And the effective_profile_snapshot column is corrupted to "not-valid-json{"
And I reload the plan from the database
Then the retrieved plan effective_profile_snapshot should be '{}'
@feature921
Scenario: Invalid JSON in effective_profile_snapshot is rejected by Plan model
When I attempt to construct a Plan with effective_profile_snapshot "not-valid-json"
Then the Plan construction should raise a validation error for effective_profile_snapshot
@feature921
Scenario: Empty string effective_profile_snapshot is rejected by Plan model
When I attempt to construct a Plan with an empty effective_profile_snapshot
Then the Plan construction should raise a validation error for effective_profile_snapshot
features/steps/action_repository_coverage_steps.py
+4 -1
View File
@@ -395,8 +395,10 @@ def step_create_referencing_plan(context: Context) -> None:
"""Insert a LifecyclePlanModel row that references the saved action."""
session = _get_session(context)
now_iso = datetime.now().isoformat()
pid = _next_ulid()
plan_model = LifecyclePlanModel(
plan_id=_next_ulid(),
plan_id=pid,
root_plan_id=pid,
action_name=str(context.action.namespaced_name),
phase="strategize",
processing_state="queued",
@@ -404,6 +406,7 @@ def step_create_referencing_plan(context: Context) -> None:
namespaced_name="local/test-plan",
namespace="local",
description="Plan referencing the action under test",
effective_profile_snapshot="{}",
created_at=now_iso,
updated_at=now_iso,
tags_json="[]",
features/steps/database_models_coverage_boost_steps.py
+10
View File
@@ -403,6 +403,7 @@ def step_plan_model_with_automation_profile(context):
)
model = LifecyclePlanModel(
plan_id=ULID_PLAN_1,
root_plan_id=ULID_PLAN_1,
action_name="local/cov-action",
namespaced_name="local/profile-plan",
namespace="local",
@@ -414,6 +415,7 @@ def step_plan_model_with_automation_profile(context):
strategy_actor="local/strategy",
execution_actor="local/executor",
automation_profile=profile_json,
effective_profile_snapshot="{}",
reusable=True,
read_only=False,
created_at=NOW_ISO,
@@ -459,6 +461,7 @@ def step_plan_model_with_sandbox_refs(context):
_ensure_action_exists(context.cb_session)
model = LifecyclePlanModel(
plan_id=ULID_PLAN_2,
root_plan_id=ULID_PLAN_2,
action_name="local/cov-action",
namespaced_name="local/sandbox-plan",
namespace="local",
@@ -470,6 +473,7 @@ def step_plan_model_with_sandbox_refs(context):
strategy_actor="local/strategy",
execution_actor="local/executor",
sandbox_refs_json=json.dumps(["sandbox-ref-1", "sandbox-ref-2"]),
effective_profile_snapshot="{}",
reusable=True,
read_only=False,
created_at=NOW_ISO,
@@ -502,6 +506,7 @@ def step_plan_model_with_validation_summary(context):
plan_id = str(ULID())
model = LifecyclePlanModel(
plan_id=plan_id,
root_plan_id=plan_id,
action_name="local/cov-action",
namespaced_name="local/validation-plan",
namespace="local",
@@ -513,6 +518,7 @@ def step_plan_model_with_validation_summary(context):
strategy_actor="local/strategy",
execution_actor="local/executor",
validation_summary_json=json.dumps(summary),
effective_profile_snapshot="{}",
reusable=True,
read_only=False,
created_at=NOW_ISO,
@@ -546,6 +552,7 @@ def step_plan_model_with_error_details(context):
plan_id = str(ULID())
model = LifecyclePlanModel(
plan_id=plan_id,
root_plan_id=plan_id,
action_name="local/cov-action",
namespaced_name="local/error-plan",
namespace="local",
@@ -558,6 +565,7 @@ def step_plan_model_with_error_details(context):
execution_actor="local/executor",
error_message="Something went wrong",
error_details_json=json.dumps(error_details),
effective_profile_snapshot="{}",
reusable=True,
read_only=False,
created_at=NOW_ISO,
@@ -590,6 +598,7 @@ def step_plan_model_with_arguments(context):
plan_id = str(ULID())
model = LifecyclePlanModel(
plan_id=plan_id,
root_plan_id=plan_id,
action_name="local/cov-action",
namespaced_name="local/args-plan",
namespace="local",
@@ -600,6 +609,7 @@ def step_plan_model_with_arguments(context):
definition_of_done="Tests pass",
strategy_actor="local/strategy",
execution_actor="local/executor",
effective_profile_snapshot="{}",
reusable=True,
read_only=False,
created_at=NOW_ISO,
features/steps/database_models_coverage_r2_steps.py
+7 -6
View File
@@ -14,6 +14,11 @@ from typing import Any
from behave import given, then, when # type: ignore[import-untyped]
from cleveragents.domain.models.core.plan import (
AutomationProfileProvenance,
AutomationProfileRef,
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
@@ -234,7 +239,7 @@ def step_check_arg_strings(context: Any) -> None:
def _base_plan_ns(
*,
processing_state: Any = "queued",
automation_profile: Any = None,
automation_profile: AutomationProfileRef | None = None,
namespaced_name: Any = None,
error_details: Any = None,
) -> SimpleNamespace:
@@ -272,6 +277,7 @@ def _base_plan_ns(
tags=[],
reusable=True,
read_only=False,
effective_profile_snapshot="{}",
)
@@ -301,11 +307,6 @@ def step_check_plan_state(context: Any) -> None:
@given("a Plan domain object with a non-null automation_profile")
def step_plan_with_automation_profile(context: Any) -> None:
from cleveragents.domain.models.core.plan import (
AutomationProfileProvenance,
AutomationProfileRef,
)
profile = AutomationProfileRef(
profile_name="strict",
provenance=AutomationProfileProvenance.ACTION,
features/steps/database_models_lifecycle_coverage_steps.py
+6 -1
View File
@@ -141,10 +141,12 @@ def _make_plan_model(
- ``tags`` -> ``tags_json``
- Added ``namespace`` column
"""
# Root plans self-reference; default when not explicitly provided.
resolved_root = root_plan_id if root_plan_id is not None else plan_id
model = LifecyclePlanModel(
plan_id=plan_id,
parent_plan_id=parent_plan_id,
root_plan_id=root_plan_id,
root_plan_id=resolved_root,
action_name=action_name,
phase=phase,
processing_state=state,
@@ -169,6 +171,7 @@ def _make_plan_model(
tags_json=tags,
reusable=reusable,
read_only=read_only,
effective_profile_snapshot="{}",
)
# Populate project links via child table
if project_names:
@@ -845,6 +848,7 @@ def _make_plan_domain(
tags=tags,
reusable=True,
read_only=False,
effective_profile_snapshot="{}",
)
@@ -991,6 +995,7 @@ def step_create_plan_domain_null_states(context):
tags=[],
reusable=True,
read_only=False,
effective_profile_snapshot="{}",
)
features/steps/database_models_new_coverage_steps.py
+1
View File
@@ -190,6 +190,7 @@ def _make_plan_object(
tags=[],
reusable=True,
read_only=False,
effective_profile_snapshot="{}",
)
features/steps/models_lifecycle_coverage_r2_steps.py
+3 -2
View File
@@ -100,7 +100,7 @@ def _make_plan_model(
model = LifecyclePlanModel(
plan_id=_ULID_PLAN,
parent_plan_id=None,
root_plan_id=None,
root_plan_id=_ULID_PLAN,
action_name=action_name or "",
namespaced_name="local/test-plan",
namespace="local",
@@ -115,7 +115,8 @@ def _make_plan_model(
apply_actor=None,
estimation_actor=None,
invariant_actor=None,
automation_profile=automation_profile,
automation_profile=automation_profile or "balanced",
effective_profile_snapshot="{}",
reusable=True,
read_only=False,
inputs_schema_json=None,
features/steps/plan_lifecycle_coverage_steps.py
+2
View File
@@ -725,6 +725,7 @@ def step_uow_add_flush(context: Context) -> None:
# Create a raw model and add it via the context
model = LifecyclePlanModel()
model.plan_id = "01TESTADD00000000000000000" # type: ignore[assignment]
model.root_plan_id = "01TESTADD00000000000000000" # type: ignore[assignment]
model.namespaced_name = "local/uow-add-test" # type: ignore[assignment]
model.namespace = "local" # type: ignore[assignment]
model.name = "uow-add-test" # type: ignore[assignment]
@@ -740,6 +741,7 @@ def step_uow_add_flush(context: Context) -> None:
model.read_only = False # type: ignore[assignment]
model.tags_json = "[]" # type: ignore[assignment]
model.sandbox_refs_json = "[]" # type: ignore[assignment]
model.effective_profile_snapshot = "{}" # type: ignore[assignment]
now = datetime.now().isoformat()
model.created_at = now # type: ignore[assignment]
model.updated_at = now # type: ignore[assignment]
features/steps/plan_phase_migration_steps.py
+4
View File
@@ -87,6 +87,7 @@ def step_insert_plan_with_phase_and_state(context: Any, phase: str, state: str)
now = _now_iso()
plan = LifecyclePlanModel()
plan.plan_id = ulid
plan.root_plan_id = ulid
plan.action_name = "local/phase-test-action"
plan.namespaced_name = "local/phase-test-plan"
plan.namespace = "local"
@@ -94,6 +95,7 @@ def step_insert_plan_with_phase_and_state(context: Any, phase: str, state: str)
plan.processing_state = state
plan.description = "Test plan"
plan.tags_json = "[]"
plan.effective_profile_snapshot = "{}"
plan.created_at = now
plan.updated_at = now
try:
@@ -155,11 +157,13 @@ def step_insert_plan_default_phase(context: Any) -> None:
# Use ORM to create plan without specifying phase — should default to 'action'
plan = LifecyclePlanModel()
plan.plan_id = ulid
plan.root_plan_id = ulid
plan.action_name = "local/phase-test-action"
plan.namespaced_name = "local/default-phase-plan"
plan.namespace = "local"
plan.description = "Default phase plan"
plan.tags_json = "[]"
plan.effective_profile_snapshot = "{}"
plan.created_at = now
plan.updated_at = now
session.add(plan)
features/steps/plans_table_schema_alignment_steps.py
+539
View File
@@ -0,0 +1,539 @@
"""Step definitions for v3_plans table schema alignment (issue #921).
Tests the effective_profile_snapshot column, root_plan_id self-referencing,
automation_profile NOT NULL default, bare-string deserialization, and
NULL root_plan_id constraint enforcement.
"""
from __future__ import annotations
from datetime import UTC, datetime
from behave import given, then, when
from behave.runner import Context
from sqlalchemy import create_engine, text
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import sessionmaker
from cleveragents.domain.models.core.action import Action, ActionState
from cleveragents.domain.models.core.plan import (
AutomationProfileProvenance,
AutomationProfileRef,
NamespacedName,
Plan,
PlanIdentity,
PlanPhase,
PlanTimestamps,
ProcessingState,
)
from cleveragents.infrastructure.database.models import (
Base,
LifecyclePlanModel,
)
from cleveragents.infrastructure.database.repositories import (
ActionRepository,
LifecyclePlanRepository,
)
_NOW = datetime(2026, 3, 19, tzinfo=UTC)
def _make_plan(
plan_id: str,
*,
parent_plan_id: str | None = None,
root_plan_id: str | None = None,
effective_profile_snapshot: str = "{}",
automation_profile: AutomationProfileRef | None = None,
action_name: str = "local/schema-test",
) -> Plan:
"""Build a Plan domain object for schema alignment tests."""
return Plan(
identity=PlanIdentity(
plan_id=plan_id,
parent_plan_id=parent_plan_id,
root_plan_id=root_plan_id,
attempt=1,
),
namespaced_name=NamespacedName(namespace="local", name="schema-test"),
action_name=action_name,
description="Schema alignment test plan",
definition_of_done="All assertions pass",
phase=PlanPhase.ACTION,
processing_state=ProcessingState.QUEUED,
automation_profile=automation_profile,
effective_profile_snapshot=effective_profile_snapshot,
strategy_actor="local/s",
execution_actor="local/e",
timestamps=PlanTimestamps(created_at=_NOW, updated_at=_NOW),
created_by="test-schema",
tags=[],
reusable=True,
read_only=False,
)
def _setup_db(context: Context) -> None:
"""Create a fresh in-memory SQLite database and attach to context."""
engine = create_engine("sqlite:///:memory:", echo=False)
Base.metadata.create_all(engine)
sm = sessionmaker(bind=engine)
session = sm()
context._sa_engine = engine
context._sa_session = session
context._sa_session_factory = lambda: session
context._sa_plan_repo = LifecyclePlanRepository(
session_factory=context._sa_session_factory,
)
context._sa_action_repo = ActionRepository(
session_factory=context._sa_session_factory,
)
@given("a schema-aligned fresh in-memory database")
def step_setup_db(context: Context) -> None:
_setup_db(context)
@given('a schema-aligned prerequisite action "{action_name}" exists')
def step_create_action(context: Context, action_name: str) -> None:
ns = NamespacedName.parse(action_name)
action = Action(
namespaced_name=ns,
description="Schema alignment test action",
definition_of_done="Done",
strategy_actor="local/s",
execution_actor="local/e",
state=ActionState.AVAILABLE,
created_at=_NOW,
updated_at=_NOW,
)
context._sa_action_repo.create(action)
context._sa_session.commit()
@given('a schema-aligned plan with ID "{plan_id}"')
def step_create_plan(context: Context, plan_id: str) -> None:
context._sa_plan = _make_plan(plan_id)
@given("the plan effective_profile_snapshot is '{snapshot}'")
def step_set_snapshot(context: Context, snapshot: str) -> None:
context._sa_plan = context._sa_plan.model_copy(
update={"effective_profile_snapshot": snapshot}
)
@given('a schema-aligned root plan with ID "{plan_id}"')
def step_create_root_plan(context: Context, plan_id: str) -> None:
# root_plan_id is None here; from_domain will set it to plan_id
context._sa_plan = _make_plan(plan_id, root_plan_id=None)
@given(
'a schema-aligned child plan with ID "{child_id}" parent "{parent_id}" root "{root_id}"'
)
def step_create_child_plan(
context: Context, child_id: str, parent_id: str, root_id: str
) -> None:
context._sa_child_plan = _make_plan(
child_id, parent_plan_id=parent_id, root_plan_id=root_id
)
@given(
'a schema-aligned grandchild plan with ID "{gc_id}" parent "{parent_id}" root "{root_id}"'
)
def step_create_grandchild_plan(
context: Context, gc_id: str, parent_id: str, root_id: str
) -> None:
context._sa_grandchild_plan = _make_plan(
gc_id, parent_plan_id=parent_id, root_plan_id=root_id
)
@given('a schema-aligned plan with ID "{plan_id}" and no automation profile')
def step_create_plan_no_profile(context: Context, plan_id: str) -> None:
context._sa_plan = _make_plan(plan_id, automation_profile=None)
@when("I persist the schema-aligned plan")
def step_persist_plan(context: Context) -> None:
context._sa_plan_repo.create(context._sa_plan)
context._sa_session.commit()
plan_id = context._sa_plan.identity.plan_id
context._sa_retrieved = context._sa_plan_repo.get(plan_id)
@when("I persist both schema-aligned plans")
def step_persist_both(context: Context) -> None:
context._sa_plan_repo.create(context._sa_plan)
context._sa_session.commit()
context._sa_plan_repo.create(context._sa_child_plan)
context._sa_session.commit()
child_id = context._sa_child_plan.identity.plan_id
context._sa_retrieved_child = context._sa_plan_repo.get(child_id)
@when("I persist root, child, and grandchild schema-aligned plans")
def step_persist_three_levels(context: Context) -> None:
context._sa_plan_repo.create(context._sa_plan)
context._sa_session.commit()
context._sa_plan_repo.create(context._sa_child_plan)
context._sa_session.commit()
context._sa_plan_repo.create(context._sa_grandchild_plan)
context._sa_session.commit()
gc_id = context._sa_grandchild_plan.identity.plan_id
context._sa_retrieved_grandchild = context._sa_plan_repo.get(gc_id)
@when("I update the plan effective_profile_snapshot to '{new_snapshot}'")
def step_update_snapshot(context: Context, new_snapshot: str) -> None:
retrieved = context._sa_retrieved
assert retrieved is not None, "Plan was not retrieved before update"
updated = retrieved.model_copy(update={"effective_profile_snapshot": new_snapshot})
context._sa_plan_repo.update(updated)
context._sa_session.commit()
plan_id = updated.identity.plan_id
context._sa_retrieved = context._sa_plan_repo.get(plan_id)
@then("the retrieved plan effective_profile_snapshot should be '{expected}'")
def step_check_snapshot(context: Context, expected: str) -> None:
retrieved = context._sa_retrieved
assert retrieved is not None, "Plan was not retrieved"
assert retrieved.effective_profile_snapshot == expected, (
f"Expected snapshot '{expected}', got '{retrieved.effective_profile_snapshot}'"
)
@then('the plan root_plan_id should equal the plan_id "{plan_id}"')
def step_check_root_self_ref(context: Context, plan_id: str) -> None:
retrieved = context._sa_retrieved
assert retrieved is not None, "Plan was not retrieved"
assert retrieved.identity.root_plan_id == plan_id, (
f"Expected root_plan_id '{plan_id}', got '{retrieved.identity.root_plan_id}'"
)
@then('the child plan root_plan_id should be "{expected}"')
def step_check_child_root(context: Context, expected: str) -> None:
child = context._sa_retrieved_child
assert child is not None, "Child plan was not retrieved"
assert child.identity.root_plan_id == expected, (
f"Expected root_plan_id '{expected}', got '{child.identity.root_plan_id}'"
)
@then('the grandchild plan root_plan_id should be "{expected}"')
def step_check_grandchild_root(context: Context, expected: str) -> None:
gc = context._sa_retrieved_grandchild
assert gc is not None, "Grandchild plan was not retrieved"
assert gc.identity.root_plan_id == expected, (
f"Expected root_plan_id '{expected}', got '{gc.identity.root_plan_id}'"
)
@then('the grandchild plan parent_plan_id should be "{expected}"')
def step_check_grandchild_parent(context: Context, expected: str) -> None:
gc = context._sa_retrieved_grandchild
assert gc is not None, "Grandchild plan was not retrieved"
assert gc.identity.parent_plan_id == expected, (
f"Expected parent_plan_id '{expected}', got '{gc.identity.parent_plan_id}'"
)
@then('the child plan parent_plan_id should be "{expected}"')
def step_check_child_parent(context: Context, expected: str) -> None:
child = context._sa_retrieved_child
assert child is not None, "Child plan was not retrieved"
assert child.identity.parent_plan_id == expected, (
f"Expected parent_plan_id '{expected}', got '{child.identity.parent_plan_id}'"
)
@then('the persisted automation_profile column should be "{expected}"')
def step_check_automation_profile_col(context: Context, expected: str) -> None:
plan_id = context._sa_plan.identity.plan_id
row = context._sa_session.execute(
text("SELECT automation_profile FROM v3_plans WHERE plan_id = :pid"),
{"pid": plan_id},
).fetchone()
assert row is not None, "Row not found in v3_plans"
assert row[0] == expected, (
f"Expected automation_profile='{expected}', got '{row[0]}'"
)
# --- M3: Bare-string / invalid-provenance automation_profile handling ---
@given('the plan automation_profile column is set to bare string "{value}"')
def step_set_bare_string_profile(context: Context, value: str) -> None:
"""Persist the plan, then overwrite the column with a bare string."""
context._sa_plan_repo.create(context._sa_plan)
context._sa_session.commit()
plan_id = context._sa_plan.identity.plan_id
context._sa_session.execute(
text("UPDATE v3_plans SET automation_profile = :val WHERE plan_id = :pid"),
{"val": value, "pid": plan_id},
)
context._sa_session.commit()
context._sa_target_plan_id = plan_id
@given("the plan automation_profile column is set to '{value}'")
def step_set_invalid_json_profile(context: Context, value: str) -> None:
"""Persist the plan, then overwrite the column with arbitrary JSON."""
context._sa_plan_repo.create(context._sa_plan)
context._sa_session.commit()
plan_id = context._sa_plan.identity.plan_id
context._sa_session.execute(
text("UPDATE v3_plans SET automation_profile = :val WHERE plan_id = :pid"),
{"val": value, "pid": plan_id},
)
context._sa_session.commit()
context._sa_target_plan_id = plan_id
@when("I load the plan from the database")
def step_load_plan(context: Context) -> None:
plan_id = context._sa_target_plan_id
context._sa_retrieved = context._sa_plan_repo.get(plan_id)
@then("the plan automation_profile ref should be None")
def step_check_profile_is_none(context: Context) -> None:
retrieved = context._sa_retrieved
assert retrieved is not None, "Plan was not retrieved"
assert retrieved.automation_profile is None, (
f"Expected automation_profile=None, got {retrieved.automation_profile!r}"
)
# --- L3: NULL root_plan_id constraint enforcement ---
@given('a schema-aligned plan with ID "{plan_id}" forced NULL root_plan_id')
def step_create_plan_null_root(context: Context, plan_id: str) -> None:
context._sa_null_root_plan_id = plan_id
@when("I attempt to persist the plan with NULL root_plan_id")
def step_persist_null_root(context: Context) -> None:
plan_id = context._sa_null_root_plan_id
now_iso = _NOW.isoformat()
try:
context._sa_session.execute(
text(
"INSERT INTO v3_plans "
"(plan_id, root_plan_id, action_name, namespaced_name, "
"namespace, phase, processing_state, attempt, description, "
"tags_json, effective_profile_snapshot, automation_profile, "
"reusable, read_only, created_at, updated_at) "
"VALUES (:pid, NULL, :act, :ns, 'local', 'action', 'queued', "
"1, 'test', '[]', '{}', 'balanced', 1, 0, :now, :now)"
),
{
"pid": plan_id,
"act": "local/schema-test",
"ns": "local/schema-test",
"now": now_iso,
},
)
context._sa_session.commit()
context._sa_integrity_error = None
except IntegrityError as exc:
context._sa_session.rollback()
context._sa_integrity_error = exc
@then("the database should reject the insert with an integrity error")
def step_check_integrity_error(context: Context) -> None:
assert context._sa_integrity_error is not None, (
"Expected IntegrityError for NULL root_plan_id, but insert succeeded"
)
# --- T2: AutomationProfileRef round-trip ---
@given(
'a schema-aligned plan "{plan_id}" with profile "{profile}" and provenance "{provenance}"'
)
def step_create_plan_with_profile(
context: Context, plan_id: str, profile: str, provenance: str
) -> None:
ref = AutomationProfileRef(
profile_name=profile,
provenance=AutomationProfileProvenance(provenance),
)
context._sa_plan = _make_plan(plan_id, automation_profile=ref)
@then('the plan automation_profile name should be "{expected}"')
def step_check_profile_name(context: Context, expected: str) -> None:
retrieved = context._sa_retrieved
assert retrieved is not None, "Plan was not retrieved"
assert retrieved.automation_profile is not None, (
"Expected automation_profile to be set, got None"
)
assert retrieved.automation_profile.profile_name == expected, (
f"Expected profile_name='{expected}', "
f"got '{retrieved.automation_profile.profile_name}'"
)
@then('the plan automation_profile provenance should be "{expected}"')
def step_check_profile_provenance(context: Context, expected: str) -> None:
retrieved = context._sa_retrieved
assert retrieved is not None, "Plan was not retrieved"
assert retrieved.automation_profile is not None, (
"Expected automation_profile to be set, got None"
)
assert retrieved.automation_profile.provenance.value == expected, (
f"Expected provenance='{expected}', "
f"got '{retrieved.automation_profile.provenance.value}'"
)
# --- TC5: NULL effective_profile_snapshot constraint enforcement ---
@given(
'a schema-aligned plan with ID "{plan_id}" forced NULL effective_profile_snapshot'
)
def step_create_plan_null_snapshot(context: Context, plan_id: str) -> None:
context._sa_null_snapshot_plan_id = plan_id
@when("I attempt to persist the plan with NULL effective_profile_snapshot")
def step_persist_null_snapshot(context: Context) -> None:
plan_id = context._sa_null_snapshot_plan_id
now_iso = _NOW.isoformat()
try:
context._sa_session.execute(
text(
"INSERT INTO v3_plans "
"(plan_id, root_plan_id, action_name, namespaced_name, "
"namespace, phase, processing_state, attempt, description, "
"tags_json, effective_profile_snapshot, automation_profile, "
"reusable, read_only, created_at, updated_at) "
"VALUES (:pid, :pid, :act, :ns, 'local', 'action', 'queued', "
"1, 'test', '[]', NULL, 'balanced', 1, 0, :now, :now)"
),
{
"pid": plan_id,
"act": "local/schema-test",
"ns": "local/schema-test",
"now": now_iso,
},
)
context._sa_session.commit()
context._sa_snapshot_integrity_error = None
except IntegrityError as exc:
context._sa_session.rollback()
context._sa_snapshot_integrity_error = exc
@then("the database should reject the snapshot insert with an integrity error")
def step_check_snapshot_integrity_error(context: Context) -> None:
assert context._sa_snapshot_integrity_error is not None, (
"Expected IntegrityError for NULL effective_profile_snapshot, "
"but insert succeeded"
)
# --- TC4: PlanIdentity model_validator self-reference resolution ---
@given('a schema-aligned plan with ID "{plan_id}" and None root_plan_id')
def step_create_plan_none_root(context: Context, plan_id: str) -> None:
context._sa_plan = _make_plan(plan_id, root_plan_id=None)
@then('the domain plan root_plan_id should already equal "{expected}"')
def step_check_domain_root_resolved(context: Context, expected: str) -> None:
plan = context._sa_plan
assert plan is not None, "Plan was not created"
assert plan.identity.root_plan_id == expected, (
f"Expected PlanIdentity.root_plan_id='{expected}' after model_validator, "
f"got '{plan.identity.root_plan_id}'"
)
@when("I convert the plan via from_domain")
def step_convert_from_domain(context: Context) -> None:
context._sa_orm_model = LifecyclePlanModel.from_domain(context._sa_plan)
@then('the ORM model root_plan_id should equal "{expected}"')
def step_check_orm_root(context: Context, expected: str) -> None:
model = context._sa_orm_model
assert model is not None, "ORM model was not created"
assert model.root_plan_id == expected, (
f"Expected ORM root_plan_id='{expected}', got '{model.root_plan_id}'"
)
# --- TC6: Invalid JSON in effective_profile_snapshot ---
@when('the effective_profile_snapshot column is corrupted to "{bad_json}"')
def step_corrupt_snapshot_column(context: Context, bad_json: str) -> None:
"""Overwrite the effective_profile_snapshot column with invalid JSON."""
plan_id = context._sa_plan.identity.plan_id
context._sa_session.execute(
text(
"UPDATE v3_plans SET effective_profile_snapshot = :val WHERE plan_id = :pid"
),
{"val": bad_json, "pid": plan_id},
)
context._sa_session.commit()
@when("I reload the plan from the database")
def step_reload_plan(context: Context) -> None:
"""Expire cached ORM state and reload from DB to pick up raw SQL changes."""
context._sa_session.expire_all()
plan_id = context._sa_plan.identity.plan_id
context._sa_retrieved = context._sa_plan_repo.get(plan_id)
@when('I attempt to construct a Plan with effective_profile_snapshot "{bad_json}"')
def step_construct_plan_bad_snapshot(context: Context, bad_json: str) -> None:
from pydantic import ValidationError as PydanticValidationError
try:
_make_plan(
"01HV000000000000000000SA21",
effective_profile_snapshot=bad_json,
)
context._sa_validation_error = None
except PydanticValidationError as exc:
context._sa_validation_error = exc
@when("I attempt to construct a Plan with an empty effective_profile_snapshot")
def step_construct_plan_empty_snapshot(context: Context) -> None:
from pydantic import ValidationError as PydanticValidationError
try:
_make_plan(
"01HV000000000000000000SA22",
effective_profile_snapshot="",
)
context._sa_validation_error = None
except PydanticValidationError as exc:
context._sa_validation_error = exc
@then(
"the Plan construction should raise a validation error "
"for effective_profile_snapshot"
)
def step_check_validation_error(context: Context) -> None:
assert context._sa_validation_error is not None, (
"Expected a ValidationError for invalid JSON in "
"effective_profile_snapshot, but Plan was created successfully"
)
features/steps/repositories_coverage_r2_steps.py
+2
View File
@@ -184,6 +184,7 @@ def _insert_v3_plan(session: Session, plan_id: str) -> None:
now_iso = datetime.now(UTC).isoformat()
plan_row = LifecyclePlanModel(
plan_id=plan_id,
root_plan_id=plan_id,
namespaced_name=f"local/test-plan-{plan_id[:8]}",
namespace="local",
action_name="local/test-action",
@@ -196,6 +197,7 @@ def _insert_v3_plan(session: Session, plan_id: str) -> None:
read_only=False,
created_by="test",
tags_json="[]",
effective_profile_snapshot="{}",
created_at=now_iso,
updated_at=now_iso,
)
features/steps/repositories_coverage_steps.py
+4 -1
View File
@@ -407,8 +407,10 @@ def step_repos_cov_create_referencing_plan(context: Context, action_name: str) -
"""Insert a LifecyclePlanModel row referencing the specified action."""
session: Session = context.rc_session
now_iso = datetime.now().isoformat()
pid = _next_ulid()
plan_model = LifecyclePlanModel(
plan_id=_next_ulid(),
plan_id=pid,
root_plan_id=pid,
action_name=action_name,
phase="strategize",
processing_state="queued",
@@ -416,6 +418,7 @@ def step_repos_cov_create_referencing_plan(context: Context, action_name: str) -
namespaced_name="local/ref-plan",
namespace="local",
description="Plan referencing action under test",
effective_profile_snapshot="{}",
created_at=now_iso,
updated_at=now_iso,
tags_json="[]",
features/steps/repositories_error_handling_coverage_steps.py
+1
View File
@@ -288,6 +288,7 @@ def _make_fake_plan() -> SimpleNamespace:
arguments={},
arguments_order=[],
invariants=[],
effective_profile_snapshot="{}",
timestamps=SimpleNamespace(
created_at=now,
updated_at=now,
robot/helper_plans_table_schema_alignment.py
+394
View File
@@ -0,0 +1,394 @@
"""Robot Framework helper for v3_plans schema alignment tests (issue #921).
Verifies effective_profile_snapshot, root_plan_id self-reference, and
automation_profile NOT NULL default.
Usage:
python robot/helper_plans_table_schema_alignment.py snapshot-roundtrip
python robot/helper_plans_table_schema_alignment.py root-self-ref
python robot/helper_plans_table_schema_alignment.py child-root-id
"""
from __future__ import annotations
import sys
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
from sqlalchemy import create_engine # noqa: E402
from sqlalchemy.orm import sessionmaker # noqa: E402
from cleveragents.domain.models.core.action import ( # noqa: E402
Action,
ActionState,
)
from cleveragents.domain.models.core.plan import ( # noqa: E402
AutomationProfileProvenance,
AutomationProfileRef,
NamespacedName,
Plan,
PlanIdentity,
PlanPhase,
PlanTimestamps,
ProcessingState,
)
from cleveragents.infrastructure.database.models import Base # noqa: E402
from cleveragents.infrastructure.database.repositories import ( # noqa: E402
ActionRepository,
LifecyclePlanRepository,
)
_NOW = datetime(2026, 3, 19, tzinfo=UTC)
_ACTION_NAME = "local/schema-test"
def _setup() -> tuple[Any, Any, Any]:
"""Create in-memory DB and return (session, plan_repo, action_repo)."""
engine = create_engine("sqlite:///:memory:", echo=False)
Base.metadata.create_all(engine)
sm = sessionmaker(bind=engine)
session = sm()
factory = lambda: session # noqa: E731
plan_repo = LifecyclePlanRepository(session_factory=factory)
action_repo = ActionRepository(session_factory=factory)
# Create prerequisite action for FK constraint
ns = NamespacedName.parse(_ACTION_NAME)
action = Action(
namespaced_name=ns,
description="Schema alignment test action",
definition_of_done="Done",
strategy_actor="local/s",
execution_actor="local/e",
state=ActionState.AVAILABLE,
created_at=_NOW,
updated_at=_NOW,
)
action_repo.create(action)
session.commit()
return session, plan_repo, action_repo
def _make_plan(
plan_id: str,
*,
parent_plan_id: str | None = None,
root_plan_id: str | None = None,
effective_profile_snapshot: str = "{}",
automation_profile: AutomationProfileRef | None = None,
) -> Plan:
return Plan(
identity=PlanIdentity(
plan_id=plan_id,
parent_plan_id=parent_plan_id,
root_plan_id=root_plan_id,
attempt=1,
),
namespaced_name=NamespacedName(namespace="local", name="schema-test"),
action_name=_ACTION_NAME,
description="Schema alignment test plan",
definition_of_done="All assertions pass",
phase=PlanPhase.ACTION,
processing_state=ProcessingState.QUEUED,
automation_profile=automation_profile,
effective_profile_snapshot=effective_profile_snapshot,
strategy_actor="local/s",
execution_actor="local/e",
timestamps=PlanTimestamps(created_at=_NOW, updated_at=_NOW),
created_by="test-schema",
tags=[],
reusable=True,
read_only=False,
)
def _cmd_snapshot_roundtrip() -> int:
"""Verify effective_profile_snapshot roundtrips through persistence."""
session, plan_repo, _ = _setup()
snapshot = '{"profile":"balanced","thresholds":{}}'
plan = _make_plan(
"01HV000000000000000000SR01",
effective_profile_snapshot=snapshot,
)
plan_repo.create(plan)
session.commit()
retrieved = plan_repo.get("01HV000000000000000000SR01")
if retrieved is None:
print("FAIL: plan not retrieved")
return 1
if retrieved.effective_profile_snapshot != snapshot:
print(
f"FAIL: snapshot mismatch: {retrieved.effective_profile_snapshot!r} "
f"!= {snapshot!r}"
)
return 1
print("schema-snapshot-ok")
return 0
def _cmd_root_self_ref() -> int:
"""Verify root plan self-references its own plan_id."""
session, plan_repo, _ = _setup()
plan_id = "01HV000000000000000000SR02"
plan = _make_plan(plan_id, root_plan_id=None)
plan_repo.create(plan)
session.commit()
retrieved = plan_repo.get(plan_id)
if retrieved is None:
print("FAIL: plan not retrieved")
return 1
if retrieved.identity.root_plan_id != plan_id:
print(
f"FAIL: root_plan_id={retrieved.identity.root_plan_id!r}, "
f"expected={plan_id!r}"
)
return 1
print("schema-root-self-ref-ok")
return 0
def _cmd_child_root_id() -> int:
"""Verify child plan carries parent's root_plan_id."""
session, plan_repo, _ = _setup()
root_id = "01HV000000000000000000SR03"
child_id = "01HV000000000000000000SR04"
root_plan = _make_plan(root_id, root_plan_id=None)
plan_repo.create(root_plan)
session.commit()
child_plan = _make_plan(child_id, parent_plan_id=root_id, root_plan_id=root_id)
plan_repo.create(child_plan)
session.commit()
retrieved = plan_repo.get(child_id)
if retrieved is None:
print("FAIL: child plan not retrieved")
return 1
if retrieved.identity.root_plan_id != root_id:
print(
f"FAIL: root_plan_id={retrieved.identity.root_plan_id!r}, "
f"expected={root_id!r}"
)
return 1
if retrieved.identity.parent_plan_id != root_id:
print(
f"FAIL: parent_plan_id={retrieved.identity.parent_plan_id!r}, "
f"expected={root_id!r}"
)
return 1
print("schema-child-root-ok")
return 0
def _cmd_grandchild_root_id() -> int:
"""Verify grandchild plan carries root ancestor's plan_id (3-level)."""
session, plan_repo, _ = _setup()
root_id = "01HV000000000000000000SR05"
child_id = "01HV000000000000000000SR06"
gc_id = "01HV000000000000000000SR07"
root_plan = _make_plan(root_id, root_plan_id=None)
plan_repo.create(root_plan)
session.commit()
child_plan = _make_plan(child_id, parent_plan_id=root_id, root_plan_id=root_id)
plan_repo.create(child_plan)
session.commit()
gc_plan = _make_plan(gc_id, parent_plan_id=child_id, root_plan_id=root_id)
plan_repo.create(gc_plan)
session.commit()
retrieved = plan_repo.get(gc_id)
if retrieved is None:
print("FAIL: grandchild plan not retrieved")
return 1
if retrieved.identity.root_plan_id != root_id:
print(
f"FAIL: root_plan_id={retrieved.identity.root_plan_id!r}, "
f"expected={root_id!r}"
)
return 1
if retrieved.identity.parent_plan_id != child_id:
print(
f"FAIL: parent_plan_id={retrieved.identity.parent_plan_id!r}, "
f"expected={child_id!r}"
)
return 1
print("schema-grandchild-root-ok")
return 0
def _cmd_automation_profile_default() -> int:
"""Verify automation_profile=None roundtrips as None via 'balanced' default."""
session, plan_repo, _ = _setup()
plan_id = "01HV000000000000000000SR08"
plan = _make_plan(plan_id) # automation_profile defaults to None
plan_repo.create(plan)
session.commit()
retrieved = plan_repo.get(plan_id)
if retrieved is None:
print("FAIL: plan not retrieved")
return 1
if retrieved.automation_profile is not None:
print(
f"FAIL: expected automation_profile=None, "
f"got {retrieved.automation_profile!r}"
)
return 1
# Verify raw column value is 'balanced'
from sqlalchemy import text
row = session.execute(
text("SELECT automation_profile FROM v3_plans WHERE plan_id = :pid"),
{"pid": plan_id},
).fetchone()
if row is None or row[0] != "balanced":
print(f"FAIL: raw column value={row[0] if row else 'N/A'}, expected='balanced'")
return 1
print("schema-automation-profile-default-ok")
return 0
def _cmd_null_root_rejected() -> int:
"""Verify NULL root_plan_id is rejected by the NOT NULL constraint."""
from sqlalchemy import text
from sqlalchemy.exc import IntegrityError as SAIntegrityError
session, _, _ = _setup()
now_iso = _NOW.isoformat()
try:
session.execute(
text(
"INSERT INTO v3_plans "
"(plan_id, root_plan_id, action_name, namespaced_name, "
"namespace, phase, processing_state, attempt, description, "
"tags_json, effective_profile_snapshot, automation_profile, "
"reusable, read_only, created_at, updated_at) "
"VALUES (:pid, NULL, :act, :ns, 'local', 'action', 'queued', "
"1, 'test', '[]', '{}', 'balanced', 1, 0, :now, :now)"
),
{
"pid": "01HV000000000000000000SR09",
"act": "local/schema-test",
"ns": "local/schema-test",
"now": now_iso,
},
)
session.commit()
print(
"FAIL: expected IntegrityError for NULL root_plan_id, but insert succeeded"
)
return 1
except SAIntegrityError:
session.rollback()
print("schema-null-root-rejected-ok")
return 0
def _cmd_automation_profile_ref_roundtrip() -> int:
"""Verify AutomationProfileRef roundtrips through persistence."""
session, plan_repo, _ = _setup()
plan_id = "01HV000000000000000000SR10"
ref = AutomationProfileRef(
profile_name="supervised",
provenance=AutomationProfileProvenance.PLAN,
)
plan = _make_plan(plan_id, automation_profile=ref)
plan_repo.create(plan)
session.commit()
retrieved = plan_repo.get(plan_id)
if retrieved is None:
print("FAIL: plan not retrieved")
return 1
if retrieved.automation_profile is None:
print("FAIL: automation_profile is None after round-trip")
return 1
if retrieved.automation_profile.profile_name != "supervised":
print(
f"FAIL: profile_name={retrieved.automation_profile.profile_name!r}, "
f"expected='supervised'"
)
return 1
if retrieved.automation_profile.provenance != AutomationProfileProvenance.PLAN:
print(
f"FAIL: provenance={retrieved.automation_profile.provenance!r}, "
f"expected='plan'"
)
return 1
print("schema-automation-profile-ref-roundtrip-ok")
return 0
def _cmd_corrupted_snapshot_fallback() -> int:
"""Verify corrupted effective_profile_snapshot falls back to '{}'."""
from sqlalchemy import text as sa_text
session, plan_repo, _ = _setup()
plan_id = "01HV000000000000000000SR11"
plan = _make_plan(
plan_id,
effective_profile_snapshot='{"profile":"balanced"}',
)
plan_repo.create(plan)
session.commit()
# Corrupt the column directly
session.execute(
sa_text(
"UPDATE v3_plans SET effective_profile_snapshot = :val WHERE plan_id = :pid"
),
{"val": "not-valid-json{", "pid": plan_id},
)
session.commit()
session.expire_all()
retrieved = plan_repo.get(plan_id)
if retrieved is None:
print("FAIL: plan not retrieved after corruption")
return 1
if retrieved.effective_profile_snapshot != "{}":
print(
f"FAIL: expected fallback '{{}}', "
f"got {retrieved.effective_profile_snapshot!r}"
)
return 1
print("schema-corrupted-snapshot-fallback-ok")
return 0
_COMMANDS: dict[str, Any] = {
"snapshot-roundtrip": _cmd_snapshot_roundtrip,
"root-self-ref": _cmd_root_self_ref,
"child-root-id": _cmd_child_root_id,
"grandchild-root-id": _cmd_grandchild_root_id,
"automation-profile-default": _cmd_automation_profile_default,
"automation-profile-ref-roundtrip": _cmd_automation_profile_ref_roundtrip,
"null-root-rejected": _cmd_null_root_rejected,
"corrupted-snapshot-fallback": _cmd_corrupted_snapshot_fallback,
}
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(2)
sys.exit(_COMMANDS[sys.argv[1]]())
robot/plans_table_schema_alignment.robot
+83
View File
@@ -0,0 +1,83 @@
*** Settings ***
Documentation Integration tests for v3_plans schema alignment (issue #921).
... Verifies effective_profile_snapshot, root_plan_id self-reference,
... and automation_profile NOT NULL default.
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER} ${CURDIR}/helper_plans_table_schema_alignment.py
*** Test Cases ***
Plan Includes Effective Profile Snapshot
[Documentation] Create a plan with an explicit profile snapshot and verify persistence.
[Tags] feature921
${result}= Run Process ${PYTHON} ${HELPER} snapshot-roundtrip 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} schema-snapshot-ok
Root Plan Self References Its Own Plan ID
[Documentation] Root plan must have root_plan_id == plan_id.
[Tags] feature921
${result}= Run Process ${PYTHON} ${HELPER} root-self-ref 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} schema-root-self-ref-ok
Child Plan Has Parent Root Plan ID
[Documentation] Child plan must carry the root ancestor's plan_id.
[Tags] feature921
${result}= Run Process ${PYTHON} ${HELPER} child-root-id 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} schema-child-root-ok
Grandchild Plan Carries Root Ancestor Plan ID
[Documentation] Grandchild plan (3-level hierarchy) must carry root ancestor's plan_id.
[Tags] feature921
${result}= Run Process ${PYTHON} ${HELPER} grandchild-root-id 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} schema-grandchild-root-ok
Automation Profile Defaults To Balanced And Roundtrips As None
[Documentation] Plan with no automation profile stores 'balanced' in DB and loads as None.
[Tags] feature921
${result}= Run Process ${PYTHON} ${HELPER} automation-profile-default 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} schema-automation-profile-default-ok
Automation Profile Ref Roundtrips Through Persistence
[Documentation] Plan with an explicit AutomationProfileRef persists and loads correctly.
[Tags] feature921
${result}= Run Process ${PYTHON} ${HELPER} automation-profile-ref-roundtrip 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} schema-automation-profile-ref-roundtrip-ok
NULL Root Plan ID Is Rejected By Database
[Documentation] Inserting a plan with NULL root_plan_id must raise IntegrityError.
[Tags] feature921
${result}= Run Process ${PYTHON} ${HELPER} null-root-rejected 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} schema-null-root-rejected-ok
Corrupted Effective Profile Snapshot Falls Back To Empty JSON
[Documentation] Corrupted effective_profile_snapshot in DB must fall back to '{}' on read.
[Tags] feature921
${result}= Run Process ${PYTHON} ${HELPER} corrupted-snapshot-fallback 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} schema-corrupted-snapshot-fallback-ok
src/cleveragents/domain/models/core/plan.py
+42
View File
@@ -56,6 +56,7 @@ Based on ``docs/specification.md`` and ADR-004 (Pydantic Validation).
from __future__ import annotations
import json
import re
from datetime import datetime
from enum import StrEnum
@@ -295,6 +296,21 @@ class PlanIdentity(BaseModel):
description="Attempt counter (increments on re-run)",
)
@model_validator(mode="after")
def resolve_root_plan_id(self) -> PlanIdentity:
"""Resolve ``root_plan_id`` to ``plan_id`` when ``None``.
Per the specification DDL, ``root_plan_id`` is ``NOT NULL``:
root plans self-reference their own ``plan_id``; child plans
carry the root ancestor's ``plan_id``. Resolving at domain
construction time keeps the domain model consistent with the
database constraint regardless of whether the object has been
persisted yet.
"""
if self.root_plan_id is None:
self.root_plan_id = self.plan_id
return self
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
@@ -707,6 +723,20 @@ class Plan(BaseModel):
description="Multi-project scope tracking and cross-project metadata",
)
# Frozen profile snapshot — JSON snapshot of the automation profile
# resolved at plan-creation time. Used for audit / reproducibility.
# NOTE: The default '{}' exists for backward compatibility with code
# paths that create Plan objects before the snapshot is populated.
# New plans SHOULD explicitly set this field to the resolved profile
# JSON at creation time; the empty default does not satisfy the spec
# intent of capturing a frozen profile for audit purposes.
effective_profile_snapshot: str = Field(
default="{}",
description=(
"Frozen JSON snapshot of the automation profile at plan creation time"
),
)
# Metadata
created_by: str | None = Field(None, description="User/session that created plan")
tags: list[str] = Field(default_factory=list, description="Tags for organization")
@@ -756,6 +786,18 @@ class Plan(BaseModel):
MAX_REVERSIONS: ClassVar[int] = 3
@field_validator("effective_profile_snapshot")
@classmethod
def validate_effective_profile_snapshot_json(cls, v: str) -> str:
"""Ensure the snapshot is well-formed JSON."""
try:
json.loads(v)
except (json.JSONDecodeError, TypeError, RecursionError) as exc:
raise ValueError(
f"effective_profile_snapshot must be valid JSON (length={len(v)})"
) from exc
return v
@model_validator(mode="after")
def validate_env_priority_requires_environment(self) -> Plan:
"""Enforce that execution_env_priority requires execution_environment.
src/cleveragents/infrastructure/database/models.py
+113 -14
View File
@@ -43,6 +43,7 @@ if TYPE_CHECKING:
from cleveragents.domain.models.core.checkpoint import Checkpoint
from cleveragents.domain.models.core.correction import CorrectionAttemptRecord
from pydantic import ValidationError
from sqlalchemy import (
JSON,
Boolean,
@@ -77,6 +78,11 @@ from cleveragents.domain.models.core import (
_logger = logging.getLogger(__name__)
# Default automation profile name. Used as the NOT NULL server_default
# for the ``automation_profile`` column and as the sentinel value that
# ``to_domain()`` treats as "no structured profile set" (returns None).
DEFAULT_AUTOMATION_PROFILE = "balanced"
# Create base class for all models
Base = declarative_base()
@@ -571,6 +577,27 @@ class LifecyclePlanModel(Base): # type: ignore[misc]
phase/state, execution context, and sandbox references.
See: ``src/cleveragents/domain/models/core/plan.py``
.. note:: **Column naming conventions (intentional deviations from spec DDL)**
The spec DDL uses ``automation_profile_name``, ``strategy_actor_name``,
``execution_actor_name``, ``estimation_actor_name``, ``invariant_actor_name``,
``state``, and ``error_traceback``. The code uses shorter forms without the
``_name`` suffix (``automation_profile``, ``strategy_actor``, etc.),
``processing_state`` (to avoid collision with SQLAlchemy internals), and
``error_details_json`` (richer structured data than a plain traceback).
The table is named ``v3_plans`` (evolution artifact) vs spec's ``plans``.
These are intentional conventions established before the spec DDL was
formalised and are kept for backward compatibility.
**Semantic note on ``automation_profile``**: The spec's
``automation_profile_name`` stores a plain profile name string.
The code's ``automation_profile`` stores either a bare name
(``"balanced"``, the NOT NULL default) or a structured JSON object
(``{"profile_name": "...", "provenance": "..."}``). The dual
format allows tracking provenance (how the profile was resolved)
alongside the name, at the cost of requiring format-aware
deserialization in ``to_domain()``.
"""
__allow_unmapped__ = True
@@ -585,8 +612,8 @@ class LifecyclePlanModel(Base): # type: ignore[misc]
)
root_plan_id = Column(
String(26),
ForeignKey("v3_plans.plan_id", ondelete="SET NULL"),
nullable=True,
ForeignKey("v3_plans.plan_id", ondelete="RESTRICT"),
nullable=False,
)
action_name = Column(
String(255),
@@ -599,6 +626,11 @@ class LifecyclePlanModel(Base): # type: ignore[misc]
namespace = Column(String(100), nullable=False)
# Lifecycle phase and processing state (spec-aligned, rebaselined in a5_005)
# NOTE: Spec DDL defaults phase to 'strategize', but the code default is
# 'action'. This is intentional — the Action phase was added as a
# pre-Strategize setup step (template instantiation) that did not exist
# in the original specification. Plans begin in Action and transition to
# Strategize via ``agents plan use``.
phase = Column(
String(20), nullable=False, default="action", server_default="action"
)
@@ -620,7 +652,15 @@ class LifecyclePlanModel(Base): # type: ignore[misc]
invariant_actor = Column(String(255), nullable=True)
# Policy / profile
automation_profile = Column(Text, nullable=True)
automation_profile = Column(
Text,
nullable=False,
default=DEFAULT_AUTOMATION_PROFILE,
server_default=DEFAULT_AUTOMATION_PROFILE,
)
effective_profile_snapshot = Column(
Text, nullable=False, default="{}", server_default="{}"
)
# Behavior
reusable = Column(Boolean, nullable=False, default=True)
@@ -723,6 +763,19 @@ class LifecyclePlanModel(Base): # type: ignore[misc]
# -- Domain conversion helpers ------------------------------------------
@staticmethod
def _serialize_automation_profile(profile_ref: Any) -> str:
"""Serialize an ``AutomationProfileRef`` to a JSON column value.
Returns the JSON representation when *profile_ref* is truthy,
otherwise the module-level default ``DEFAULT_AUTOMATION_PROFILE``.
Centralises the serialisation logic used by both ``from_domain()``
and ``LifecyclePlanRepository.update()``.
"""
if profile_ref is not None:
return json.dumps(profile_ref.model_dump(mode="json"))
return DEFAULT_AUTOMATION_PROFILE
@staticmethod
def _parse_iso(value: str | None) -> datetime | None:
"""Parse an ISO-8601 string to ``datetime``, returning ``None`` for empty."""
@@ -830,14 +883,57 @@ class LifecyclePlanModel(Base): # type: ignore[misc]
else:
arguments_dict[arg_name] = None
# Deserialize automation profile
# Deserialize automation profile. The column is NOT NULL with
# server_default 'balanced'. Values are either a JSON object
# (``{"profile_name": "...", "provenance": "..."}``), or a bare
# profile-name string (e.g. ``"balanced"``). Bare strings and
# the default ``"balanced"`` value are treated as unparseable
# and result in ``automation_profile_ref = None``.
automation_profile_ref: AutomationProfileRef | None = None
if self.automation_profile is not None:
profile_raw: dict[str, str] = json.loads(cast(str, self.automation_profile))
automation_profile_ref = AutomationProfileRef(
profile_name=profile_raw["profile_name"],
provenance=AutomationProfileProvenance(profile_raw["provenance"]),
raw_ap = cast(str, self.automation_profile)
if raw_ap and raw_ap != DEFAULT_AUTOMATION_PROFILE:
try:
profile_raw: dict[str, str] = json.loads(raw_ap)
automation_profile_ref = AutomationProfileRef(
profile_name=profile_raw["profile_name"],
provenance=AutomationProfileProvenance(profile_raw["provenance"]),
)
except (
json.JSONDecodeError,
KeyError,
ValueError,
ValidationError,
RecursionError,
):
# Bare profile name string, invalid provenance enum,
# Pydantic validation failure, or deeply nested JSON
# — treat as unparseable.
_logger.warning(
"Unparseable automation_profile for plan %s (length=%d)",
self.plan_id,
len(raw_ap),
)
automation_profile_ref = None
# Validate effective_profile_snapshot. The column is NOT NULL
# with server_default '{}', but corrupted rows may contain invalid
# JSON. Mirror the defensive pattern used for automation_profile
# above to prevent a single corrupted row from crashing reads.
raw_snapshot = (
cast(str, self.effective_profile_snapshot)
if self.effective_profile_snapshot is not None
else "{}"
)
try:
json.loads(raw_snapshot)
except (json.JSONDecodeError, TypeError, RecursionError):
_logger.warning(
"Unparseable effective_profile_snapshot for plan %s (length=%d); "
"falling back to '{}'",
self.plan_id,
len(raw_snapshot),
)
raw_snapshot = "{}"
# Deserialize sandbox refs
sandbox_refs_list: list[str] = json.loads(
@@ -867,7 +963,7 @@ class LifecyclePlanModel(Base): # type: ignore[misc]
identity=PlanIdentity(
plan_id=cast(str, self.plan_id),
parent_plan_id=cast("str | None", self.parent_plan_id),
root_plan_id=cast("str | None", self.root_plan_id),
root_plan_id=cast(str, self.root_plan_id),
attempt=cast(int, self.attempt),
),
namespaced_name=NamespacedName.parse(cast(str, self.namespaced_name)),
@@ -877,6 +973,7 @@ class LifecyclePlanModel(Base): # type: ignore[misc]
phase=phase_enum,
processing_state=state_enum,
automation_profile=automation_profile_ref,
effective_profile_snapshot=raw_snapshot,
strategy_actor=cast("str | None", self.strategy_actor),
execution_actor=cast("str | None", self.execution_actor),
review_actor=cast("str | None", self.review_actor),
@@ -950,12 +1047,13 @@ class LifecyclePlanModel(Base): # type: ignore[misc]
)
# Serialize automation profile
automation_profile_json = (
json.dumps(plan.automation_profile.model_dump(mode="json"))
if plan.automation_profile
else None
automation_profile_json = cls._serialize_automation_profile(
plan.automation_profile,
)
# root_plan_id is resolved by PlanIdentity's model_validator
# (None → plan_id for root plans); always non-None here.
tags_json = json.dumps(plan.tags)
resolved_action = action_name or plan.action_name or ""
@@ -980,6 +1078,7 @@ class LifecyclePlanModel(Base): # type: ignore[misc]
estimation_actor=getattr(plan, "estimation_actor", None),
invariant_actor=getattr(plan, "invariant_actor", None),
automation_profile=automation_profile_json,
effective_profile_snapshot=plan.effective_profile_snapshot,
reusable=plan.reusable,
read_only=plan.read_only,
execution_environment=plan.execution_environment,
src/cleveragents/infrastructure/database/repositories.py
+4 -6
View File
@@ -1371,13 +1371,11 @@ class LifecyclePlanRepository:
row.estimation_actor = getattr(plan, "estimation_actor", None) # type: ignore[assignment]
row.invariant_actor = getattr(plan, "invariant_actor", None) # type: ignore[assignment]
# Serialize automation profile
automation_profile_json = (
json.dumps(plan.automation_profile.model_dump(mode="json"))
if plan.automation_profile
else None
# Serialize automation profile — column is NOT NULL, default 'balanced'
row.automation_profile = LifecyclePlanModel._serialize_automation_profile( # type: ignore[assignment]
plan.automation_profile,
)
row.automation_profile = automation_profile_json # type: ignore[assignment]
row.effective_profile_snapshot = plan.effective_profile_snapshot # type: ignore[assignment]
row.error_message = plan.error_message # type: ignore[assignment]
row.error_details_json = ( # type: ignore[assignment]