feat(uow): wire lifecycle repositories
CI / lint (pull_request) Successful in 16s
CI / typecheck (pull_request) Successful in 30s
CI / security (pull_request) Successful in 22s
CI / quality (pull_request) Successful in 15s
CI / integration_tests (pull_request) Successful in 4m45s
CI / build (pull_request) Successful in 16s
CI / unit_tests (pull_request) Successful in 10m35s
CI / coverage (pull_request) Successful in 7m16s
CI / docker (pull_request) Successful in 38s
CI / typecheck (push) Waiting to run
CI / lint (push) Waiting to run
CI / security (push) Waiting to run
CI / quality (push) Waiting to run
CI / unit_tests (push) Waiting to run
CI / integration_tests (push) Waiting to run
CI / coverage (push) Blocked by required conditions
CI / build (push) Waiting to run
CI / docker (push) Blocked by required conditions

This commit was merged in pull request #67.
This commit is contained in:
Jeffrey Phillips Freeman
2026-02-15 09:04:17 +00:00
parent 5691858d46
commit ae40bb24c0
8 changed files with 862 additions and 16 deletions
+198
View File
@@ -0,0 +1,198 @@
"""ASV benchmarks for UnitOfWork lifecycle repository wiring.
Measures overhead of creating UnitOfWorkContext and accessing
actions / lifecycle_plans repositories.
"""
from __future__ import annotations
from datetime import datetime
from sqlalchemy import create_engine, event
from sqlalchemy.orm import Session, sessionmaker
from cleveragents.domain.models.core.action import Action, ActionState
from cleveragents.domain.models.core.plan import (
AutomationLevel,
NamespacedName,
PlanIdentity,
PlanPhase,
PlanTimestamps,
ProcessingState,
)
from cleveragents.domain.models.core.plan import Plan as V3Plan
from cleveragents.infrastructure.database.models import Base
from cleveragents.infrastructure.database.unit_of_work import UnitOfWorkContext
_CB32 = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
_bench_counter = 5000
def _bench_ulid() -> str:
global _bench_counter
_bench_counter += 1
n = _bench_counter
suffix = ""
for _ in range(8):
suffix = _CB32[n % 32] + suffix
n //= 32
return f"01HGZ6FE0AQDYTR4BX{suffix}"
def _make_bench_action(name: str = "local/bench-uow-action") -> Action:
parts = name.split("/", 1)
namespace = parts[0] if len(parts) == 2 else "local"
short_name = parts[1] if len(parts) == 2 else parts[0]
now = datetime.now()
return Action(
namespaced_name=NamespacedName(namespace=namespace, name=short_name),
description=f"Bench action {short_name}",
long_description=None,
definition_of_done="Done",
strategy_actor="local/s",
execution_actor="local/e",
estimation_actor=None,
review_actor=None,
arguments=[],
reusable=True,
read_only=False,
state=ActionState("available"),
created_at=now,
updated_at=now,
created_by=None,
tags=[],
)
def _make_bench_plan(
plan_id: str | None = None,
action_name: str = "local/bench-uow-action",
) -> V3Plan:
now = datetime.now()
return V3Plan(
identity=PlanIdentity(plan_id=plan_id or _bench_ulid(), attempt=1),
namespaced_name=NamespacedName(namespace="local", name="bench-uow-plan"),
action_name=action_name,
description="Bench plan",
definition_of_done="Done",
strategy_actor="local/s",
execution_actor="local/e",
phase=PlanPhase("action"),
processing_state=ProcessingState("queued"),
automation_level=AutomationLevel("manual"),
timestamps=PlanTimestamps(created_at=now, updated_at=now),
project_links=[],
arguments={},
arguments_order=[],
invariants=[],
reusable=True,
read_only=False,
created_by=None,
tags=[],
)
class TimeUowContextCreation:
"""Benchmark UnitOfWorkContext instantiation."""
timeout = 30.0
def setup(self) -> None:
self.engine = create_engine(
"sqlite:///:memory:",
echo=False,
future=True,
connect_args={"check_same_thread": False},
)
@event.listens_for(self.engine, "connect")
def _fk(dbapi_conn, _rec): # type: ignore[no-untyped-def]
cursor = dbapi_conn.cursor()
cursor.execute("PRAGMA foreign_keys=ON")
cursor.close()
Base.metadata.create_all(self.engine)
self.sf: sessionmaker[Session] = sessionmaker(
bind=self.engine,
expire_on_commit=False,
autoflush=False,
autocommit=False,
class_=Session,
)
def time_create_context(self) -> None:
session = self.sf()
_ = UnitOfWorkContext(session)
session.close()
def time_access_actions_repo(self) -> None:
session = self.sf()
ctx = UnitOfWorkContext(session)
_ = ctx.actions
session.close()
def time_access_lifecycle_plans_repo(self) -> None:
session = self.sf()
ctx = UnitOfWorkContext(session)
_ = ctx.lifecycle_plans
session.close()
class TimeUowActionPlanRoundTrip:
"""Benchmark creating action + plan via UoW and retrieving."""
timeout = 30.0
def setup(self) -> None:
self.engine = create_engine(
"sqlite:///:memory:",
echo=False,
future=True,
connect_args={"check_same_thread": False},
)
@event.listens_for(self.engine, "connect")
def _fk(dbapi_conn, _rec): # type: ignore[no-untyped-def]
cursor = dbapi_conn.cursor()
cursor.execute("PRAGMA foreign_keys=ON")
cursor.close()
Base.metadata.create_all(self.engine)
self.sf: sessionmaker[Session] = sessionmaker(
bind=self.engine,
expire_on_commit=False,
autoflush=False,
autocommit=False,
class_=Session,
)
# Seed an action for plan creation benchmarks
session = self.sf()
ctx = UnitOfWorkContext(session)
ctx.actions.create(_make_bench_action("local/bench-uow-action"))
session.commit()
session.close()
def time_create_action_via_uow(self) -> None:
session = self.sf()
ctx = UnitOfWorkContext(session)
uid = _bench_ulid()
ctx.actions.create(_make_bench_action(f"local/bench-action-{uid}"))
session.commit()
session.close()
def time_create_plan_via_uow(self) -> None:
session = self.sf()
ctx = UnitOfWorkContext(session)
ctx.lifecycle_plans.create(_make_bench_plan())
session.commit()
session.close()
def time_create_action_and_plan_via_uow(self) -> None:
session = self.sf()
ctx = UnitOfWorkContext(session)
uid = _bench_ulid()
action_name = f"local/bench-combo-{uid}"
ctx.actions.create(_make_bench_action(action_name))
ctx.lifecycle_plans.create(_make_bench_plan(action_name=action_name))
session.commit()
session.close()
+48
View File
@@ -83,3 +83,51 @@ See `ActionRepository` class documentation in `repositories.py` for full method
|-----------|--------|-------------|
| `DuplicateActionError` | `DatabaseError` | Raised when creating an action with a name that already exists. |
| `ActionInUseError` | `BusinessRuleViolation` | Raised when deleting an action still referenced by plans. |
## UnitOfWork Lifecycle Usage
**Module**: `src/cleveragents/infrastructure/database/unit_of_work.py`
The `UnitOfWork` class provides transactional access to all repositories. The `UnitOfWorkContext` exposes both legacy and v3 lifecycle repositories within a single transaction boundary.
### Available Repositories on UnitOfWorkContext
| Property | Type | Description |
|----------|------|-------------|
| `actions` | `ActionRepository` | V3 lifecycle action persistence. |
| `lifecycle_plans` | `LifecyclePlanRepository` | V3 lifecycle plan persistence. |
| `projects` | `ProjectRepository` | Legacy project persistence. |
| `plans` | `PlanRepository` | Legacy plan persistence (deprecated; use `lifecycle_plans`). |
| `contexts` | `ContextRepository` | Context item persistence. |
| `changes` | `ChangeRepository` | Change persistence. |
| `debug_attempts` | `DebugAttemptRepository` | Debug attempt persistence. |
| `actors` | `ActorRepository` | Actor persistence. |
### Transaction Pattern
```python
from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
uow = UnitOfWork(database_url="sqlite:///app.db")
# All operations within the context manager are committed atomically
with uow.transaction() as ctx:
ctx.actions.create(action)
ctx.lifecycle_plans.create(plan)
# Both persisted on successful exit; rolled back on exception
```
### Session-Factory Bridging
The v3 repositories (`ActionRepository`, `LifecyclePlanRepository`) use a session-factory pattern (`Callable[[], Session]`), while legacy repositories take a `Session` directly. `UnitOfWorkContext` bridges both patterns by providing a `_session_factory()` method that returns the transaction's session, ensuring all repositories operate within the same transaction boundary.
### Cross-Repository Atomicity
Creating an action and a plan in the same transaction guarantees that either both persist or neither does:
```python
with uow.transaction() as ctx:
ctx.actions.create(action)
ctx.lifecycle_plans.create(plan)
# If plan creation fails, the action is also rolled back
```
+353
View File
@@ -0,0 +1,353 @@
"""Step definitions for UnitOfWork lifecycle repository wiring.
Targets ``UnitOfWork`` and ``UnitOfWorkContext`` in
``src/cleveragents/infrastructure/database/unit_of_work.py``.
"""
from __future__ import annotations
from datetime import datetime
from typing import Any
from behave import given, then, when
from behave.runner import Context
from sqlalchemy import create_engine, event
from sqlalchemy.orm import Session, sessionmaker
from cleveragents.domain.models.core.action import Action, ActionState
from cleveragents.domain.models.core.plan import (
AutomationLevel,
NamespacedName,
PlanIdentity,
PlanPhase,
PlanTimestamps,
ProcessingState,
)
from cleveragents.domain.models.core.plan import Plan as V3Plan
from cleveragents.infrastructure.database.models import Base
from cleveragents.infrastructure.database.repositories import (
ActionRepository,
LifecyclePlanRepository,
PlanRepository,
)
from cleveragents.infrastructure.database.unit_of_work import (
UnitOfWorkContext,
)
def _make_uow_action(name: str = "local/uow-action") -> Action:
"""Create a minimal Action domain object for testing."""
parts = name.split("/", 1)
namespace = parts[0] if len(parts) == 2 else "local"
short_name = parts[1] if len(parts) == 2 else parts[0]
now = datetime.now()
return Action(
namespaced_name=NamespacedName(namespace=namespace, name=short_name),
description=f"UoW test action {short_name}",
long_description=None,
definition_of_done=f"Verify {short_name} completes",
strategy_actor="local/strategist",
execution_actor="local/executor",
estimation_actor=None,
review_actor=None,
arguments=[],
reusable=True,
read_only=False,
state=ActionState("available"),
created_at=now,
updated_at=now,
created_by=None,
tags=[],
)
def _make_uow_plan(
plan_id: str,
action_name: str = "local/uow-action",
) -> V3Plan:
"""Create a minimal Plan domain object for testing."""
now = datetime.now()
return V3Plan(
identity=PlanIdentity(plan_id=plan_id, attempt=1),
namespaced_name=NamespacedName(namespace="local", name="uow-plan"),
action_name=action_name,
description="UoW test plan",
definition_of_done="Verify UoW plan completes",
strategy_actor="local/strategist",
execution_actor="local/executor",
phase=PlanPhase("action"),
processing_state=ProcessingState("queued"),
automation_level=AutomationLevel("manual"),
timestamps=PlanTimestamps(
created_at=now,
updated_at=now,
),
project_links=[],
arguments={},
arguments_order=[],
invariants=[],
reusable=True,
read_only=False,
created_by=None,
tags=[],
)
# ---------------------------------------------------------------------------
# Background
# ---------------------------------------------------------------------------
@given("a UnitOfWork backed by an in-memory database for lifecycle tests")
def step_uow_lifecycle_background(context: Context) -> None:
"""Set up an in-memory database with schema for UoW tests.
We bypass the migration runner and use ``Base.metadata.create_all``
directly since this is a unit test with an in-memory DB.
"""
engine = create_engine(
"sqlite:///:memory:",
echo=False,
future=True,
connect_args={"check_same_thread": False},
)
# Enable FK enforcement for SQLite
@event.listens_for(engine, "connect")
def _set_sqlite_fk(dbapi_conn: Any, _rec: Any) -> None:
cursor = dbapi_conn.cursor()
cursor.execute("PRAGMA foreign_keys=ON")
cursor.close()
Base.metadata.create_all(engine)
sf: sessionmaker[Session] = sessionmaker(
bind=engine,
expire_on_commit=False,
autoflush=False,
autocommit=False,
class_=Session,
)
# Store on context for reuse
context.uow_engine = engine
context.uow_session_factory = sf
# ---------------------------------------------------------------------------
# Repository exposure
# ---------------------------------------------------------------------------
@when("I access the actions repository from the UoW context")
def step_access_actions_repo(context: Context) -> None:
session = context.uow_session_factory()
ctx = UnitOfWorkContext(session)
context.uow_ctx_result = ctx.actions
session.close()
@then("the actions repository should be an ActionRepository instance")
def step_check_actions_repo_type(context: Context) -> None:
assert isinstance(context.uow_ctx_result, ActionRepository), (
f"Expected ActionRepository, got {type(context.uow_ctx_result)}"
)
@when("I access the lifecycle_plans repository from the UoW context")
def step_access_lifecycle_plans_repo(context: Context) -> None:
session = context.uow_session_factory()
ctx = UnitOfWorkContext(session)
context.uow_ctx_result = ctx.lifecycle_plans
session.close()
@then("the lifecycle_plans repository should be a LifecyclePlanRepository instance")
def step_check_lifecycle_plans_repo_type(context: Context) -> None:
assert isinstance(context.uow_ctx_result, LifecyclePlanRepository), (
f"Expected LifecyclePlanRepository, got {type(context.uow_ctx_result)}"
)
# ---------------------------------------------------------------------------
# CRUD via UoW
# ---------------------------------------------------------------------------
@when('I create an action "{name}" via the UoW transaction')
def step_create_action_via_uow(context: Context, name: str) -> None:
session = context.uow_session_factory()
try:
ctx = UnitOfWorkContext(session)
action = _make_uow_action(name)
ctx.actions.create(action)
session.commit()
except Exception:
session.rollback()
raise
finally:
session.close()
@then('the action "{name}" should be retrievable from the actions repository')
def step_check_action_retrievable(context: Context, name: str) -> None:
session = context.uow_session_factory()
try:
ctx = UnitOfWorkContext(session)
action = ctx.actions.get_by_name(name)
assert action is not None, f"Action {name} not found"
finally:
session.close()
@given('an action "{name}" exists via the UoW')
def step_ensure_action_exists(context: Context, name: str) -> None:
session = context.uow_session_factory()
try:
ctx = UnitOfWorkContext(session)
existing = ctx.actions.get_by_name(name)
if existing is None:
action = _make_uow_action(name)
ctx.actions.create(action)
session.commit()
except Exception:
session.rollback()
raise
finally:
session.close()
@when('I create a lifecycle plan "{plan_id}" via the UoW transaction')
def step_create_plan_via_uow(context: Context, plan_id: str) -> None:
session = context.uow_session_factory()
try:
ctx = UnitOfWorkContext(session)
plan = _make_uow_plan(plan_id, action_name="local/uow-plan-action")
ctx.lifecycle_plans.create(plan)
session.commit()
except Exception:
session.rollback()
raise
finally:
session.close()
@then('the plan "{plan_id}" should be retrievable from the lifecycle_plans repository')
def step_check_plan_retrievable(context: Context, plan_id: str) -> None:
session = context.uow_session_factory()
try:
ctx = UnitOfWorkContext(session)
plan = ctx.lifecycle_plans.get(plan_id)
assert plan is not None, f"Plan {plan_id} not found"
finally:
session.close()
# ---------------------------------------------------------------------------
# Cross-repository commit
# ---------------------------------------------------------------------------
@when(
'I create an action "{action_name}" and a plan "{plan_id}" in a single UoW transaction'
)
def step_create_action_and_plan(
context: Context, action_name: str, plan_id: str
) -> None:
session = context.uow_session_factory()
try:
ctx = UnitOfWorkContext(session)
action = _make_uow_action(action_name)
ctx.actions.create(action)
plan = _make_uow_plan(plan_id, action_name=action_name)
ctx.lifecycle_plans.create(plan)
session.commit()
except Exception:
session.rollback()
raise
finally:
session.close()
@then('the action "{name}" should be retrievable in a new transaction')
def step_check_action_new_txn(context: Context, name: str) -> None:
session = context.uow_session_factory()
try:
ctx = UnitOfWorkContext(session)
action = ctx.actions.get_by_name(name)
assert action is not None, f"Action {name} not found in new transaction"
finally:
session.close()
@then('the plan "{plan_id}" should be retrievable in a new transaction')
def step_check_plan_new_txn(context: Context, plan_id: str) -> None:
session = context.uow_session_factory()
try:
ctx = UnitOfWorkContext(session)
plan = ctx.lifecycle_plans.get(plan_id)
assert plan is not None, f"Plan {plan_id} not found in new transaction"
finally:
session.close()
# ---------------------------------------------------------------------------
# Rollback
# ---------------------------------------------------------------------------
@when('I create an action "{action_name}" and a plan "{plan_id}" but force a rollback')
def step_create_and_rollback(context: Context, action_name: str, plan_id: str) -> None:
session = context.uow_session_factory()
try:
ctx = UnitOfWorkContext(session)
action = _make_uow_action(action_name)
ctx.actions.create(action)
plan = _make_uow_plan(plan_id, action_name=action_name)
ctx.lifecycle_plans.create(plan)
# Force rollback instead of commit
session.rollback()
finally:
session.close()
@then('the action "{name}" should not exist in a new transaction')
def step_check_action_not_exists(context: Context, name: str) -> None:
session = context.uow_session_factory()
try:
ctx = UnitOfWorkContext(session)
action = ctx.actions.get_by_name(name)
assert action is None, f"Action {name} should not exist after rollback"
finally:
session.close()
@then('the plan "{plan_id}" should not exist in a new transaction')
def step_check_plan_not_exists(context: Context, plan_id: str) -> None:
session = context.uow_session_factory()
try:
ctx = UnitOfWorkContext(session)
plan = ctx.lifecycle_plans.get(plan_id)
assert plan is None, f"Plan {plan_id} should not exist after rollback"
finally:
session.close()
# ---------------------------------------------------------------------------
# Legacy accessor
# ---------------------------------------------------------------------------
@when("I access the legacy plans repository from the UoW context")
def step_access_legacy_plans_repo(context: Context) -> None:
session = context.uow_session_factory()
ctx = UnitOfWorkContext(session)
context.uow_ctx_result = ctx.plans
session.close()
@then("the legacy plans repository should be a PlanRepository instance")
def step_check_legacy_plans_repo_type(context: Context) -> None:
assert isinstance(context.uow_ctx_result, PlanRepository), (
f"Expected PlanRepository, got {type(context.uow_ctx_result)}"
)
+66
View File
@@ -0,0 +1,66 @@
@phase1 @domain @unit_of_work @uow_lifecycle
Feature: UnitOfWork lifecycle repository wiring
As a system operator managing v3 lifecycle entities
I want the UnitOfWork to expose actions and lifecycle_plans repositories
So that transactional persistence works across both repositories atomically
Background:
Given a UnitOfWork backed by an in-memory database for lifecycle tests
# ---------------------------------------------------------------------------
# Repository exposure
# ---------------------------------------------------------------------------
@uow_actions_repo
Scenario: UoW exposes actions repository
When I access the actions repository from the UoW context
Then the actions repository should be an ActionRepository instance
@uow_lifecycle_plans_repo
Scenario: UoW exposes lifecycle_plans repository
When I access the lifecycle_plans repository from the UoW context
Then the lifecycle_plans repository should be a LifecyclePlanRepository instance
# ---------------------------------------------------------------------------
# CRUD via UoW
# ---------------------------------------------------------------------------
@uow_create_action
Scenario: Create action via UoW and retrieve it
When I create an action "local/uow-action" via the UoW transaction
Then the action "local/uow-action" should be retrievable from the actions repository
@uow_create_plan
Scenario: Create plan via UoW and retrieve it
Given an action "local/uow-plan-action" exists via the UoW
When I create a lifecycle plan "01HGZ6FE0AQDYTR4BX00000E01" via the UoW transaction
Then the plan "01HGZ6FE0AQDYTR4BX00000E01" should be retrievable from the lifecycle_plans repository
# ---------------------------------------------------------------------------
# Cross-repository commit
# ---------------------------------------------------------------------------
@uow_commit_persists
Scenario: UoW commit persists changes across repositories
When I create an action "local/commit-action" and a plan "01HGZ6FE0AQDYTR4BX00000E02" in a single UoW transaction
Then the action "local/commit-action" should be retrievable in a new transaction
And the plan "01HGZ6FE0AQDYTR4BX00000E02" should be retrievable in a new transaction
# ---------------------------------------------------------------------------
# Rollback
# ---------------------------------------------------------------------------
@uow_rollback
Scenario: UoW rollback discards changes across repositories
When I create an action "local/rollback-action" and a plan "01HGZ6FE0AQDYTR4BX00000E03" but force a rollback
Then the action "local/rollback-action" should not exist in a new transaction
And the plan "01HGZ6FE0AQDYTR4BX00000E03" should not exist in a new transaction
# ---------------------------------------------------------------------------
# Legacy accessor still works
# ---------------------------------------------------------------------------
@uow_legacy_plans
Scenario: Legacy plans accessor still returns PlanRepository
When I access the legacy plans repository from the UoW context
Then the legacy plans repository should be a PlanRepository instance
+15 -15
View File
@@ -1675,24 +1675,24 @@ No standalone Q0-Advanced commits planned. Advanced QA enhancements are bundled
- [ ] Git [Jeff]: `git branch -d feature/m1-plan-repo`
- [X] Quality [Jeff]: Verify coverage >=97% via `nox -s coverage_report`. If coverage is <97% then review the current unit test coverage report at `build/coverage.xml` and use it to write new Behave based unit tests to improve code coverage. Specifically, write Behave style unit tests that are descriptively named and specifically improves coverage on whichever file has the most uncovered lines by writing tests that will target the uncovered lines in the report. Once that is done rerun `nox -s coverage_report` to verify all tests pass and coverage is above >=97%, if not repeat this task as many times as is needed until coverage reaches >=97%. Done: Day 7, February 15, 2026
- [ ] **COMMIT (Owner: Jeff | Group: A5.epsilon | Branch: feature/m1-uow-lifecycle | Planned: Day 9 | Expected: Day 11) - Commit message: "feat(uow): wire lifecycle repositories"**
- [ ] Git [Jeff]: `git checkout master`
- [ ] Git [Jeff]: `git pull origin master`
- [ ] Git [Jeff]: `git checkout -b feature/m1-uow-lifecycle`
- [ ] Git [Jeff]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
- [ ] Code [Jeff]: Update `UnitOfWork` to expose `actions` + `lifecycle_plans` repositories and remove legacy plan repository accessors.
- [ ] Code [Jeff]: Update `UnitOfWorkContext` to include new repos and remove legacy plan/change repositories from default access.
- [ ] Docs [Jeff]: Update `docs/reference/repositories.md` with UoW lifecycle usage patterns.
- [ ] Tests (Behave) [Jeff]: Add scenarios verifying UoW exposes action + lifecycle plan repositories.
- [ ] Tests (Robot) [Jeff]: Add Robot smoke test that creates an action + plan via UoW.
- [ ] Tests (ASV) [Jeff]: Add `benchmarks/uow_lifecycle_bench.py` for UoW overhead baseline.
- [ ] Quality [Jeff]: Run `nox` (all default sessions, including benchmark), fix any errors if needed ensuring nox passes.
- [ ] Git [Jeff]: `git add .`
- [ ] Git [Jeff]: `git commit -m "feat(uow): wire lifecycle repositories"`
- [X] **COMMIT (Owner: Jeff | Group: A5.epsilon | Branch: feature/m1-uow-lifecycle | Planned: Day 9 | Done: Day 7, February 15, 2026) - Commit message: "feat(uow): wire lifecycle repositories"**
- [X] Git [Jeff]: `git checkout master` Done: Day 7, February 15, 2026
- [X] Git [Jeff]: `git pull origin master` Done: Day 7, February 15, 2026
- [X] Git [Jeff]: `git checkout -b feature/m1-uow-lifecycle` Done: Day 7, February 15, 2026
- [X] Git [Jeff]: `git fetch origin && git merge origin/master` (run before final tests and before commit) Done: Day 7, February 15, 2026
- [X] Code [Jeff]: Update `UnitOfWork` to expose `actions` + `lifecycle_plans` repositories and remove legacy plan repository accessors. Done: Day 7, February 15, 2026
- [X] Code [Jeff]: Update `UnitOfWorkContext` to include new repos and remove legacy plan/change repositories from default access. Done: Day 7, February 15, 2026
- [X] Docs [Jeff]: Update `docs/reference/repositories.md` with UoW lifecycle usage patterns. Done: Day 7, February 15, 2026
- [X] Tests (Behave) [Jeff]: Add scenarios verifying UoW exposes action + lifecycle plan repositories. Done: Day 7, February 15, 2026
- [X] Tests (Robot) [Jeff]: Add Robot smoke test that creates an action + plan via UoW. Done: Day 7, February 15, 2026
- [X] Tests (ASV) [Jeff]: Add `benchmarks/uow_lifecycle_bench.py` for UoW overhead baseline. Done: Day 7, February 15, 2026
- [X] Quality [Jeff]: Run `nox` (all default sessions, including benchmark), fix any errors if needed ensuring nox passes. Done: Day 7, February 15, 2026
- [X] Git [Jeff]: `git add .` Done: Day 7, February 15, 2026
- [X] Git [Jeff]: `git commit -m "feat(uow): wire lifecycle repositories"` Done: Day 7, February 15, 2026
- [ ] Forgejo PR [Jeff]: Open PR from `feature/m1-uow-lifecycle` to `master` with description "Wire lifecycle repositories into UnitOfWork and remove legacy plan repo access.".
- [ ] Git [Jeff]: `git checkout master`
- [ ] Git [Jeff]: `git branch -d feature/m1-uow-lifecycle`
- [ ] Quality [Jeff]: Verify coverage >=97% via `nox -s coverage_report`. If coverage is <97% then review the current unit test coverage report at `build/coverage.xml` and use it to write new Behave based unit tests to improve code coverage. Specifically, write Behave style unit tests that are descriptively named and specifically improves coverage on whichever file has the most uncovered lines by writing tests that will target the uncovered lines in the report. Once that is done rerun `nox -s coverage_report` to verify all tests pass and coverage is above >=97%, if not repeat this task as many times as is needed until coverage reaches >=97%.
- [X] Quality [Jeff]: Verify coverage >=97% via `nox -s coverage_report`. If coverage is <97% then review the current unit test coverage report at `build/coverage.xml` and use it to write new Behave based unit tests to improve code coverage. Specifically, write Behave style unit tests that are descriptively named and specifically improves coverage on whichever file has the most uncovered lines by writing tests that will target the uncovered lines in the report. Once that is done rerun `nox -s coverage_report` to verify all tests pass and coverage is above >=97%, if not repeat this task as many times as is needed until coverage reaches >=97%. Done: Day 7, February 15, 2026
- [ ] **COMMIT (Owner: Jeff | Group: A5.zeta | Branch: feature/m1-lifecycle-persist | Planned: Day 10 | Expected: Day 12) - Commit message: "feat(service): persist plan lifecycle via repositories"**
- [ ] Git [Jeff]: `git checkout master`
+128
View File
@@ -0,0 +1,128 @@
"""Helper script for uow_lifecycle.robot smoke test."""
from __future__ import annotations
import sys
sys.path.insert(0, "src")
sys.path.insert(0, ".")
from datetime import datetime
from sqlalchemy import create_engine, event
from sqlalchemy.orm import Session, sessionmaker
from cleveragents.domain.models.core.action import Action, ActionState
from cleveragents.domain.models.core.plan import (
AutomationLevel,
NamespacedName,
PlanIdentity,
PlanPhase,
PlanTimestamps,
ProcessingState,
)
from cleveragents.domain.models.core.plan import Plan as V3Plan
from cleveragents.infrastructure.database.models import Base
from cleveragents.infrastructure.database.repositories import (
ActionRepository,
LifecyclePlanRepository,
)
from cleveragents.infrastructure.database.unit_of_work import UnitOfWorkContext
def main() -> None:
engine = create_engine(
"sqlite:///:memory:",
echo=False,
future=True,
connect_args={"check_same_thread": False},
)
@event.listens_for(engine, "connect")
def _fk(dbapi_conn, _rec): # type: ignore[no-untyped-def]
cursor = dbapi_conn.cursor()
cursor.execute("PRAGMA foreign_keys=ON")
cursor.close()
Base.metadata.create_all(engine)
sf: sessionmaker[Session] = sessionmaker(
bind=engine,
expire_on_commit=False,
autoflush=False,
autocommit=False,
class_=Session,
)
# --- Test 1: create action + plan via UoW ---
session = sf()
ctx = UnitOfWorkContext(session)
assert isinstance(ctx.actions, ActionRepository), "actions not ActionRepository"
assert isinstance(ctx.lifecycle_plans, LifecyclePlanRepository), (
"lifecycle_plans not LifecyclePlanRepository"
)
now = datetime.now()
action = Action(
namespaced_name=NamespacedName(namespace="local", name="robot-uow-action"),
description="Robot UoW test action",
long_description=None,
definition_of_done="Done",
strategy_actor="local/s",
execution_actor="local/e",
estimation_actor=None,
review_actor=None,
arguments=[],
reusable=True,
read_only=False,
state=ActionState("available"),
created_at=now,
updated_at=now,
created_by=None,
tags=[],
)
ctx.actions.create(action)
plan = V3Plan(
identity=PlanIdentity(plan_id="01HGZ6FE0AQDYTR4BX00000R01", attempt=1),
namespaced_name=NamespacedName(namespace="local", name="robot-uow-plan"),
action_name="local/robot-uow-action",
description="Robot UoW test plan",
definition_of_done="Done",
strategy_actor="local/s",
execution_actor="local/e",
phase=PlanPhase("action"),
processing_state=ProcessingState("queued"),
automation_level=AutomationLevel("manual"),
timestamps=PlanTimestamps(created_at=now, updated_at=now),
project_links=[],
arguments={},
arguments_order=[],
invariants=[],
reusable=True,
read_only=False,
created_by=None,
tags=[],
)
ctx.lifecycle_plans.create(plan)
session.commit()
session.close()
# --- Test 2: verify retrieval in new session ---
session2 = sf()
ctx2 = UnitOfWorkContext(session2)
found_action = ctx2.actions.get_by_name("local/robot-uow-action")
assert found_action is not None, "Action not found after commit"
found_plan = ctx2.lifecycle_plans.get("01HGZ6FE0AQDYTR4BX00000R01")
assert found_plan is not None, "Plan not found after commit"
session2.close()
print("PASS: uow_lifecycle smoke test")
if __name__ == "__main__":
try:
main()
except Exception as exc:
print(f"FAIL: {exc}", file=sys.stderr)
sys.exit(1)
+17
View File
@@ -0,0 +1,17 @@
*** Settings ***
Documentation Smoke test for UnitOfWork lifecycle repository wiring
Library Process
Library OperatingSystem
Library String
*** Variables ***
${PYTHON} python
*** Test Cases ***
UoW Lifecycle Action And Plan Via Helper Script
[Documentation] Create action + plan via UoW, verify retrieval in new session
${result}= Run Process ${PYTHON} ${CURDIR}/helper_uow_lifecycle.py
... stdout=STDOUT stderr=STDOUT timeout=30s
Log ${result.stdout}
Should Contain ${result.stdout} PASS: uow_lifecycle smoke test
Should Be Equal As Integers ${result.rc} 0
@@ -14,10 +14,12 @@ from sqlalchemy.orm import Session, sessionmaker
from cleveragents.infrastructure.database.engine_cache import MEMORY_ENGINES
from cleveragents.infrastructure.database.repositories import (
ActionRepository,
ActorRepository,
ChangeRepository,
ContextRepository,
DebugAttemptRepository,
LifecyclePlanRepository,
PlanRepository,
ProjectRepository,
)
@@ -180,6 +182,12 @@ class UnitOfWorkContext:
self._changes: ChangeRepository | None = None
self._debug_attempts: DebugAttemptRepository | None = None
self._actors: ActorRepository | None = None
self._actions: ActionRepository | None = None
self._lifecycle_plans: LifecyclePlanRepository | None = None
def _session_factory(self) -> Session:
"""Return the transaction's session for factory-pattern repositories."""
return self._session
@property
def projects(self) -> ProjectRepository:
@@ -190,11 +198,39 @@ class UnitOfWorkContext:
@property
def plans(self) -> PlanRepository:
"""Get plan repository for this transaction."""
"""Get legacy plan repository for this transaction.
.. deprecated::
Use :attr:`lifecycle_plans` for v3 lifecycle plans.
This accessor targets the legacy ``plans`` table and will
be removed in a future release.
"""
if self._plans is None:
self._plans = PlanRepository(self._session)
return self._plans
@property
def actions(self) -> ActionRepository:
"""Get v3 action repository for this transaction.
Uses the session-factory pattern required by ActionRepository.
"""
if self._actions is None:
self._actions = ActionRepository(session_factory=self._session_factory)
return self._actions
@property
def lifecycle_plans(self) -> LifecyclePlanRepository:
"""Get v3 lifecycle plan repository for this transaction.
Uses the session-factory pattern required by LifecyclePlanRepository.
"""
if self._lifecycle_plans is None:
self._lifecycle_plans = LifecyclePlanRepository(
session_factory=self._session_factory,
)
return self._lifecycle_plans
@property
def contexts(self) -> ContextRepository:
"""Get context repository for this transaction."""