fix: InvariantService persistence across CLI invocations (Bug #8573)

Delegate global/project scope CRUD to DI-wired repository instead of ad-hoc sessions; switch CLI service resolution to DI container wiring.

.forgejo/workflows/benchmark-scheduled.yml

@@ -15,9 +15,6 @@ on:
                 required: false
                 default: "false"
 
-vars:
-    docker_prefix: "http://harbor.cleverthis.com/docker/"
-
 env:
     UV_VERSION: "0.8.0"
     PYTHON_VERSION: "3.13"
@@ -29,7 +26,7 @@ jobs:
         runs-on: docker
         timeout-minutes: 120
         container:
-            image: ${{vars.docker_prefix}}python:3.13-slim
+            image: python:3.13-slim
         steps:
             - name: Install system dependencies
               run: |
@@ -114,7 +111,7 @@ jobs:
         runs-on: docker
         timeout-minutes: 180
         container:
-            image: ${{vars.docker_prefix}}python:3.13-slim
+            image: python:3.13-slim
         steps:
             - name: Install system dependencies
               run: |

.forgejo/workflows/ci.yml

@@ -6,9 +6,6 @@ on:
     pull_request:
         branches: [master, develop*]
 
-vars:
-    docker_prefix: "http://harbor.cleverthis.com/docker/"
-
 env:
     UV_VERSION: "0.8.0"
     PYTHON_VERSION: "3.13"
@@ -18,7 +15,7 @@ jobs:
     lint:
         runs-on: docker
         container:
-            image: ${{vars.docker_prefix}}python:3.13-slim
+            image: python:3.13-slim
         steps:
             - name: Install Node.js (required by actions/checkout)
               run: |
@@ -62,7 +59,7 @@ jobs:
     typecheck:
         runs-on: docker
         container:
-            image: ${{vars.docker_prefix}}python:3.13-slim
+            image: python:3.13-slim
         steps:
             - name: Install Node.js (required by actions/checkout)
               run: |
@@ -100,7 +97,7 @@ jobs:
     security:
         runs-on: docker
         container:
-            image: ${{vars.docker_prefix}}python:3.13-slim
+            image: python:3.13-slim
         steps:
             - name: Install Node.js (required by actions/checkout)
               run: |
@@ -144,7 +141,7 @@ jobs:
     quality:
         runs-on: docker
         container:
-            image: ${{vars.docker_prefix}}python:3.13-slim
+            image: python:3.13-slim
         steps:
             - name: Install Node.js (required by actions/checkout)
               run: |
@@ -182,7 +179,7 @@ jobs:
     unit_tests:
         runs-on: docker
         container:
-            image: ${{vars.docker_prefix}}python:3.13-slim
+            image: python:3.13-slim
         steps:
             - name: Install system dependencies (nodejs for checkout, git for merge tests, curl/tar for Helm)
               run: |
@@ -233,7 +230,7 @@ jobs:
     integration_tests:
         runs-on: docker
         container:
-            image: ${{vars.docker_prefix}}python:3.13-slim
+            image: python:3.13-slim
         steps:
             - name: Install system dependencies (nodejs for checkout, git for integration tests, curl/tar for Helm)
               run: |
@@ -288,60 +285,10 @@ jobs:
                   path: build/nox-integration-tests-output.log
                   retention-days: 30
 
-    e2e_tests:
-        runs-on: docker
-        timeout-minutes: 45
-        container:
-            image: ${{vars.docker_prefix}}python:3.13-slim
-        steps:
-            - name: Install system dependencies (nodejs for checkout, git for E2E tests)
-              run: |
-                  apt-get update && apt-get install -y -qq nodejs git && rm -rf /var/lib/apt/lists/*
-
-            - uses: actions/checkout@v4
-
-            - name: Install uv and nox
-              run: |
-                  pip install -q uv==${{ env.UV_VERSION }} nox
-
-            - name: Cache uv packages
-              uses: actions/cache@v3
-              with:
-                  path: ~/.cache/uv
-                  key: uv-${{ hashFiles('pyproject.toml') }}
-                  restore-keys: |
-                      uv-
-
-            - name: Run E2E tests via nox
-              run: |
-                  mkdir -p build
-                  nox -s e2e_tests 2>&1 | tee build/nox-e2e-tests-output.log
-              env:
-                  NOX_DEFAULT_VENV_BACKEND: uv
-                  # Run E2E suites in parallel via pabot.  4 workers keeps
-                  # wall-clock time well under the 45-minute timeout while
-                  # staying within the memory budget of the docker runner.
-                  TEST_PROCESSES: "4"
-                  ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
-                  OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
-                  GOOGLE_API_KEY: ${{ secrets.GOOGLE_API_KEY }}
-
-            - name: Upload E2E tests log artifact
-              if: failure()
-              uses: actions/upload-artifact@v3
-              with:
-                  name: ci-logs-e2e-tests
-                  path: |
-                      build/nox-e2e-tests-output.log
-                      build/reports/robot-e2e/
-                  retention-days: 30
-
     coverage:
         runs-on: docker
         container:
-            image: ${{vars.docker_prefix}}python:3.13-slim
-        # unit_tests is included so coverage only runs after tests pass,
-        # preventing misleading results when tests are still in-flight or failing.
+            image: python:3.13-slim
         needs: [lint, typecheck, security, quality, unit_tests]
         steps:
             - name: Install system dependencies (nodejs for checkout, git for merge tests)
@@ -416,7 +363,7 @@ jobs:
     build:
         runs-on: docker
         container:
-            image: ${{vars.docker_prefix}}python:3.13-slim
+            image: python:3.13-slim
         steps:
             - name: Install Node.js (required by actions/checkout)
               run: |
@@ -472,7 +419,7 @@ jobs:
     helm:
         runs-on: docker
         container:
-            image: ${{vars.docker_prefix}}python:3.13-slim
+            image: python:3.13-slim
         steps:
             - name: Install system dependencies (nodejs for checkout, curl for Helm and kubeconform)
               run: |
@@ -537,7 +484,7 @@ jobs:
         # config (name/email) was set — both are required for any push operation.
         runs-on: docker
         container:
-            image: ${{vars.docker_prefix}}python:3.13-slim
+            image: python:3.13-slim
         steps:
             - name: Install system dependencies (nodejs for checkout, git for push validation)
               run: |
@@ -610,10 +557,10 @@ jobs:
                   echo "=== Push access smoke-test passed ==="
     status-check:
         if: always()
-        needs: [lint, typecheck, security, quality, unit_tests, integration_tests, e2e_tests, coverage, build, docker, helm, push-validation]
+        needs: [lint, typecheck, security, quality, unit_tests, integration_tests, coverage, build, docker, helm, push-validation]
         runs-on: docker
         container:
-            image: ${{vars.docker_prefix}}python:3.13-slim
+            image: python:3.13-slim
         steps:
             - name: Check required job results
               run: |
@@ -623,7 +570,6 @@ jobs:
                   echo "quality: ${{ needs.quality.result }}"
                   echo "unit_tests: ${{ needs.unit_tests.result }}"
                   echo "integration_tests: ${{ needs.integration_tests.result }}"
-                  echo "e2e_tests: ${{ needs.e2e_tests.result }}"
                   echo "coverage: ${{ needs.coverage.result }}"
                   echo "build: ${{ needs.build.result }}"
                   echo "docker: ${{ needs.docker.result }}"
@@ -636,7 +582,6 @@ jobs:
                      [ "${{ needs.quality.result }}" != "success" ] || \
                      [ "${{ needs.unit_tests.result }}" != "success" ] || \
                      [ "${{ needs.integration_tests.result }}" != "success" ] || \
-                     [ "${{ needs.e2e_tests.result }}" != "success" ] || \
                      [ "${{ needs.coverage.result }}" != "success" ] || \
                      [ "${{ needs.build.result }}" != "success" ] || \
                      [ "${{ needs.docker.result }}" != "success" ] || \

.forgejo/workflows/master.yml

@@ -3,11 +3,6 @@ name: CI
 on:
     push:
         branches: [master, develop]
-    pull_request:
-        branches: [master, develop]
-
-vars:
-    docker_prefix: "http://harbor.cleverthis.com/docker/"
 
 env:
     UV_VERSION: "0.8.0"
@@ -15,10 +10,67 @@ env:
     NOX_DEFAULT_VENV_BACKEND: "uv"
 
 jobs:
+    benchmark-regression:
+        if: forgejo.event_name != 'pull_request'
+        runs-on: docker-benchmark
+        container:
+            image: python:3.13-slim
+
+        steps:
+            - name: Install system dependencies (nodejs for checkout, git for merge tests)
+              run: |
+                  apt-get update && apt-get install -y -qq nodejs git && rm -rf /var/lib/apt/lists/*
+
+            - name: Checkout full history
+              uses: actions/checkout@v4
+              with:
+                  fetch-depth: 0
+
+            - name: Compute base commit
+              id: hash
+              run: |
+                  git fetch origin "${{ forgejo.base_ref }}" --depth=200
+                  BASE_SHA=$(git merge-base HEAD "origin/${{ forgejo.base_ref }}")
+                  echo "ASV_BASE_SHA=${BASE_SHA}" >> $FORGEJO_OUTPUT
+
+            - name: Install dependencies
+              run: |
+                  python -m pip install -U pip
+                  python -m pip install asv virtualenv uv==${{ env.UV_VERSION }} nox
+
+            - name: Restore prior ASV benchmarks
+              env:
+                  AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
+                  AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
+                  AWS_DEFAULT_REGION: ${{ secrets.AWS_DEFAULT_REGION }}
+                  ASV_S3_BUCKET: ${{ secrets.ASV_S3_BUCKET }}
+              run: |
+                  python -m pip install awscli
+                  mkdir -p build/asv/results
+                  aws s3 sync "s3://${ASV_S3_BUCKET}/asv/results" build/asv/results --delete || true
+
+            - name: Run asv continuous via nox
+              env:
+                  ASV_BASE_SHA:  ${{ steps.hash.outputs.ASV_BASE_SHA }}
+              run: |
+                  nox -s benchmark_regression
+
+            - name: Archive the results
+              run: |
+                  tar cf /tmp/asv-results.tar build/asv/results build/asv/html
+
+            - name: Upload benchmark artifacts
+              if: always()
+              uses: actions/upload-artifact@v3
+              with:
+                  name: asv-results-pr
+                  path: /tmp/asv-results.tar
+                  retention-days: 30
+
     benchmark-publish:
         if: forgejo.event_name == 'push'
         runs-on: docker-benchmark
-        container: ${{vars.docker_prefix}}python:3.13-slim
+        container: python:3.13-slim
         steps:
             - name: Install system dependencies (nodejs for checkout, git for merge tests)
               run: |
@@ -71,56 +123,51 @@ jobs:
                   name: asv-results-pr
                   path: /tmp/asv-results.tar
                   retention-days: 30
-
-    benchmark-regression:
-        # Runs benchmark regression on pull requests to detect performance regressions.
-        # This job is informational only — it is NOT listed in status-check's required
-        # needs, so a benchmark regression does not block PR merges.
-        if: forgejo.event_name == 'pull_request'
-        runs-on: docker-benchmark
-        container: ${{vars.docker_prefix}}python:3.13-slim
+    e2e_tests:
+        if: forgejo.event_name == 'push'
+        runs-on: docker
+        timeout-minutes: 45
+        container:
+            image: python:3.13-slim
         steps:
-            - name: Install system dependencies (nodejs for checkout, git for ASV)
+            - name: Install system dependencies (nodejs for checkout, git for E2E tests)
               run: |
-                  apt-get update && apt-get install -y -qq nodejs git curl && rm -rf /var/lib/apt/lists/*
+                  apt-get update && apt-get install -y -qq nodejs git && rm -rf /var/lib/apt/lists/*
 
-            - name: Checkout full history
-              uses: actions/checkout@v4
+            - uses: actions/checkout@v4
+
+            - name: Install uv and nox
+              run: |
+                  pip install -q uv==${{ env.UV_VERSION }} nox
+
+            - name: Cache uv packages
+              uses: actions/cache@v3
               with:
-                  fetch-depth: 0
+                  path: ~/.cache/uv
+                  key: uv-${{ hashFiles('pyproject.toml') }}
+                  restore-keys: |
+                      uv-
 
-            - name: Install dependencies
-              run: |
-                  python -m pip install -U pip
-                  python -m pip install asv virtualenv uv==${{ env.UV_VERSION }} nox
-
-            - name: Sync prior benchmark results from S3
-              env:
-                  AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
-                  AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
-                  AWS_DEFAULT_REGION: ${{ secrets.AWS_DEFAULT_REGION }}
-                  ASV_S3_BUCKET: ${{ secrets.ASV_S3_BUCKET }}
-              run: |
-                  if [ -n "${AWS_ACCESS_KEY_ID}" ] && [ -n "${ASV_S3_BUCKET}" ]; then
-                    python -m pip install awscli
-                    mkdir -p build/asv/results
-                    aws s3 sync "s3://${ASV_S3_BUCKET}/asv/results/" build/asv/results/ || echo "No existing results to sync"
-                  else
-                    echo "Skipping S3 sync - AWS credentials not configured"
-                  fi
-
-            - name: Run benchmark regression via nox
-              env:
-                  NOX_DEFAULT_VENV_BACKEND: uv
-                  ASV_BASE_SHA: master
+            - name: Run E2E tests via nox
               run: |
                   mkdir -p build
-                  nox -s benchmark_regression 2>&1 | tee build/nox-benchmark-regression-output.log
+                  nox -s e2e_tests 2>&1 | tee build/nox-e2e-tests-output.log
+              env:
+                  NOX_DEFAULT_VENV_BACKEND: uv
+                  # Run E2E suites in parallel via pabot.  4 workers keeps
+                  # wall-clock time well under the 45-minute timeout while
+                  # staying within the memory budget of the docker runner.
+                  TEST_PROCESSES: "4"
+                  ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
+                  OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
+                  GOOGLE_API_KEY: ${{ secrets.GOOGLE_API_KEY }}
 
-            - name: Upload benchmark regression log artifact
-              if: always()
+            - name: Upload E2E tests log artifact
+              if: failure()
               uses: actions/upload-artifact@v3
               with:
-                  name: benchmark-regression-logs
-                  path: build/nox-benchmark-regression-output.log
-                  retention-days: 30
+                  name: ci-logs-e2e-tests
+                  path: |
+                      build/nox-e2e-tests-output.log
+                      build/reports/robot-e2e/
+                  retention-days: 30

.forgejo/workflows/nightly-quality.yml

@@ -7,9 +7,6 @@ on:
   workflow_dispatch:
     # Allow manual trigger for testing
 
-vars:
-  docker_prefix: "http://harbor.cleverthis.com/docker/"
-
 env:
   UV_VERSION: "0.8.0"
   PYTHON_VERSION: "3.13"
@@ -19,7 +16,7 @@ jobs:
   full-quality-suite:
     runs-on: docker
     container:
-      image: ${{vars.docker_prefix}}python:3.13-slim
+      image: python:3.13-slim
     steps:
       - name: Install system dependencies (git for merge tests)
         run: |

.forgejo/workflows/release.yml

@@ -5,9 +5,6 @@ on:
         tags:
             - "v*"
 
-vars:
-    docker_prefix: "http://harbor.cleverthis.com/docker/"
-
 env:
     UV_VERSION: "0.8.0"
     PYTHON_VERSION: "3.13"
@@ -17,7 +14,7 @@ jobs:
     build-wheel:
         runs-on: docker
         container:
-            image: ${{vars.docker_prefix}}python:3.13-slim
+            image: python:3.13-slim
         steps:
             - name: Install Node.js (required by actions/checkout)
               run: |
@@ -88,7 +85,7 @@ jobs:
     create-release:
         runs-on: docker
         container:
-            image: ${{vars.docker_prefix}}python:3.13-slim
+            image: python:3.13-slim
         needs: [build-wheel, build-docker]
         steps:
             - name: Install system dependencies

.opencode/agents/pr-review-worker.md

@@ -136,6 +136,18 @@ Entered when `review_type` is `ci_flag`. This is a lightweight review that only
 
 Entered when `review_type` is `re_review`. The PR previously received `REQUEST_CHANGES` feedback and the author has since pushed new commits. Your job is to verify the feedback was adequately addressed AND conduct a full review of the current state.
 
+0. **Post review-started notification.** POST `{forgejo_url}/api/v1/repos/{forgejo_owner}/{forgejo_repo}/issues/{pr_number}/comments` with body:
+   ```
+   > Starting review for PR #{pr_number} (`re_review`)...
+   >
+   > Previous feedback was addressed — verifying changes and conducting full re-review. This may take a few minutes.
+
+   ---
+   Automated by CleverAgents Bot
+   Supervisor: PR Review | Agent: pr-review-worker
+   ```
+   Use authentication: `Authorization: token {forgejo_pat}`. Do NOT wait for a response before continuing — fire and move on.
+
 1. **Read the PR.** GET `{forgejo_url}/api/v1/repos/{forgejo_owner}/{forgejo_repo}/pulls/{pr_number}` — read title, body, labels, milestone, and linked issues.
 
 2. **Read all linked issues.** Parse `Closes #N`, `Fixes #N`, and `Refs #N` patterns from the PR body. For each referenced issue, GET `{forgejo_url}/api/v1/repos/{forgejo_owner}/{forgejo_repo}/issues/{issue_number}` — read title, body, labels, and acceptance criteria.
@@ -176,6 +188,18 @@ Entered when `review_type` is `re_review`. The PR previously received `REQUEST_C
 
 Entered when `review_type` is `first_review`. The PR has no active review feedback — this is a fresh evaluation from scratch.
 
+0. **Post review-started notification.** POST `{forgejo_url}/api/v1/repos/{forgejo_owner}/{forgejo_repo}/issues/{pr_number}/comments` with body:
+   ```
+   > Starting review for PR #{pr_number} (`first_review`)...
+   >
+   > Conducting fresh evaluation against the full review checklist. This may take a few minutes.
+
+   ---
+   Automated by CleverAgents Bot
+   Supervisor: PR Review | Agent: pr-review-worker
+   ```
+   Use authentication: `Authorization: token {forgejo_pat}`. Do NOT wait for a response before continuing — fire and move on.
+
 1. **Read the PR.** Same as Re-Review Mode step 1.
 
 2. **Read all linked issues.** Same as Re-Review Mode step 2.

.opencode/agents/task-implementor.md

@@ -121,16 +121,28 @@ This is where actual implementation happens. Choose the appropriate procedure ba
 
 1. **Read the issue.** GET `{forgejo_url}/api/v1/repos/{forgejo_owner}/{forgejo_repo}/issues/{work_number}` — read title, body, labels, milestone, and metadata section (branch name, commit message format). Paginate all comments to understand full context and any subtask structure.
 
-2. **Determine branch name.** Extract the branch name from the issue's Metadata section if present. If absent, derive one: `feature/issue-{work_number}-{kebab-slug-of-title}`.
+2. **Post work-started notification.** POST `{forgejo_url}/api/v1/repos/{forgejo_owner}/{forgejo_repo}/issues/{work_number}/comments` with body:
+   ```
+   > Starting implementation for issue #{work_number}: {issue_title} (`issue_impl`)...
+   >
+   > Reading requirements and setting up development environment. This may take several minutes.
 
-3. **Create isolated clone.** Call `git-isolator-util` with `create_branch: true`, `base_branch: master`, and the determined `branch_name` (see Subagents section for prompt template).
+   ---
+   Automated by CleverAgents Bot
+   Supervisor: Implementation | Agent: task-implementor
+   ```
+   Use authentication: `Authorization: token {forgejo_pat}`. Replace `{issue_title}` with the title read in step 1. Do NOT wait for a response before continuing — fire and move on.
 
-4. **Implement the code.** Load the `cleverthis-guidelines` skill for CONTRIBUTING.md rules and follow them strictly. Key rules:
+3. **Determine branch name.** Extract the branch name from the issue's Metadata section if present. If absent, derive one: `feature/issue-{work_number}-{kebab-slug-of-title}`.
+
+4. **Create isolated clone.** Call `git-isolator-util` with `create_branch: true`, `base_branch: master`, and the determined `branch_name` (see Subagents section for prompt template).
+
+5. **Implement the code.** Load the `cleverthis-guidelines` skill for CONTRIBUTING.md rules and follow them strictly. Key rules:
    - Source in `src/cleveragents/`, Behave unit tests in `features/`, Robot Framework integration/e2e tests in `robot/`
    - Full static typing throughout — no `# type: ignore`
    - All commands via `nox` — never invoke `pip`, `pytest`, `behave`, or `robot` directly
 
-5. **Run quality gates in order:**
+6. **Run quality gates in order:**
    ```bash
    nox -e lint
    nox -e typecheck
@@ -140,11 +152,11 @@ This is where actual implementation happens. Choose the appropriate procedure ba
    nox -e coverage_report
    ```
 
-6. **Fix any failures.** If a gate fails, fix the code and re-run the failing gate (and any that follow it). Repeat until all gates pass. Do not move forward with failing gates.
+7. **Fix any failures.** If a gate fails, fix the code and re-run the failing gate (and any that follow it). Repeat until all gates pass. Do not move forward with failing gates.
 
-7. **Commit.** Call `git-commit-util` with `commit_and_push` operation. The first line of the commit message must match the issue's Metadata section exactly (see Subagents section).
+8. **Commit.** Call `git-commit-util` with `commit_and_push` operation. The first line of the commit message must match the issue's Metadata section exactly (see Subagents section).
 
-8. **Create PR.** POST `{forgejo_url}/api/v1/repos/{forgejo_owner}/{forgejo_repo}/pulls` with:
+9. **Create PR.** POST `{forgejo_url}/api/v1/repos/{forgejo_owner}/{forgejo_repo}/pulls` with:
    - `title`: taken from the issue title or the commit message first line
    - `body`: description of changes + `Closes #{work_number}` + dependency link (`This PR blocks issue #{work_number}`)
    - `base`: `master`
@@ -152,35 +164,47 @@ This is where actual implementation happens. Choose the appropriate procedure ba
    - `milestone` (if set on the issue): same milestone ID
    - Use PAT authentication: `Authorization: token {forgejo_pat}`
 
-9. **Post attempt comment** on the issue (see "Attempt Comments" section below).
+10. **Post attempt comment** on the issue (see "Attempt Comments" section below).
 
-10. **Clean up.** `rm -rf {repo_dir}`
+11. **Clean up.** `rm -rf {repo_dir}`
 
-11. **Exit.**
+12. **Exit.**
 
 #### Procedure: `pr_fix` (PR Fix)
 
 1. **Read the PR.** GET `{forgejo_url}/api/v1/repos/{forgejo_owner}/{forgejo_repo}/pulls/{work_number}` — read description, head branch, head SHA, and CI state. Set `branch_name` to the PR's head branch.
 
-2. **Read all reviews.** GET `{forgejo_url}/api/v1/repos/{forgejo_owner}/{forgejo_repo}/pulls/{work_number}/reviews?limit=50&page=N` — paginate fully. For any review in `REQUEST_CHANGES` state, GET its comments to understand the specific feedback.
+2. **Post work-started notification.** POST `{forgejo_url}/api/v1/repos/{forgejo_owner}/{forgejo_repo}/issues/{work_number}/comments` with body:
+   ```
+   > Starting fix for PR #{work_number}: {pr_title} (`pr_fix`)...
+   >
+   > Addressing review feedback and CI failures. This may take several minutes.
 
-3. **Read all PR comments.** GET `{forgejo_url}/api/v1/repos/{forgejo_owner}/{forgejo_repo}/issues/{work_number}/comments?limit=50&page=N` — paginate fully.
+   ---
+   Automated by CleverAgents Bot
+   Supervisor: Implementation | Agent: task-implementor
+   ```
+   Use authentication: `Authorization: token {forgejo_pat}`. Replace `{pr_title}` with the title read in step 1. Do NOT wait for a response before continuing — fire and move on.
 
-4. **Fetch CI failure details.** GET `{forgejo_url}/api/v1/repos/{forgejo_owner}/{forgejo_repo}/commits/{head_sha}/statuses?limit=50&page=N` — paginate fully. For each failing status that has a `target_url`, webfetch that URL to retrieve the failure logs.
+3. **Read all reviews.** GET `{forgejo_url}/api/v1/repos/{forgejo_owner}/{forgejo_repo}/pulls/{work_number}/reviews?limit=50&page=N` — paginate fully. For any review in `REQUEST_CHANGES` state, GET its comments to understand the specific feedback.
 
-5. **Create isolated clone.** Call `git-isolator-util` with `create_branch: false` and `branch: {branch_name}` (the PR's head branch, resolved in step 1).
+4. **Read all PR comments.** GET `{forgejo_url}/api/v1/repos/{forgejo_owner}/{forgejo_repo}/issues/{work_number}/comments?limit=50&page=N` — paginate fully.
 
-6. **Fix the issues.** Address all CI failures and all unresolved reviewer feedback. Never partially address reviewer comments — every `REQUEST_CHANGES` concern must be fully resolved.
+5. **Fetch CI failure details.** GET `{forgejo_url}/api/v1/repos/{forgejo_owner}/{forgejo_repo}/commits/{head_sha}/statuses?limit=50&page=N` — paginate fully. For each failing status that has a `target_url`, webfetch that URL to retrieve the failure logs.
 
-7. **Run quality gates locally** (same 6 gates as above). All must pass before pushing. Fix and re-run as many times as needed.
+6. **Create isolated clone.** Call `git-isolator-util` with `create_branch: false` and `branch: {branch_name}` (the PR's head branch, resolved in step 1).
 
-8. **Commit and push.** Call `git-commit-util` with `force_push_with_lease` operation (see Subagents section).
+7. **Fix the issues.** Address all CI failures and all unresolved reviewer feedback. Never partially address reviewer comments — every `REQUEST_CHANGES` concern must be fully resolved.
 
-9. **Post attempt comment** on the PR (see "Attempt Comments" section below).
+8. **Run quality gates locally** (same 6 gates as above). All must pass before pushing. Fix and re-run as many times as needed.
 
-10. **Clean up.** `rm -rf {repo_dir}`
+9. **Commit and push.** Call `git-commit-util` with `force_push_with_lease` operation (see Subagents section).
 
-11. **Exit.**
+10. **Post attempt comment** on the PR (see "Attempt Comments" section below).
+
+11. **Clean up.** `rm -rf {repo_dir}`
+
+12. **Exit.**
 
 ### Attempt Comments
 

CHANGELOG.md

@@ -5,6 +5,14 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ## [Unreleased]
 
+- **`task-implementor` posts work-started notification comments** (#11031): Both
+  the `issue_impl` and `pr_fix` procedures now post an informational "work
+  started" comment to the Forgejo issue/PR before beginning implementation.
+  The comment includes the issue/PR title, procedure type, and expected
+  duration. Posted asynchronously ("fire and move on") so it does not block
+  the workflow. Step numbering in both procedures has been re-numbered to
+  accommodate the new step.
+
 - **`agents session tell` invokes real LLM orchestrator actor** (#5784): Replaced the
   M3 echo-stub with real actor invocation via `SessionWorkflow`, routing through
   `LangChainSessionCaller` → `ToolCallingRuntime.run_tool_loop()`.  The user prompt
@@ -26,9 +34,52 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ### Added
 
+- **`pr-review-worker` review-started notification** (#11028): The `first_review`
+  and `re_review` modes now post a "review started" notification comment to the
+  PR at the beginning of the review, giving PR authors immediate visibility
+  that a review is in progress. Posted asynchronously so it does not block
+  the workflow.
+
 - **Plan Rollback Command** (#8557): Implemented `agents plan rollback <plan-id> [<checkpoint-id>]` for checkpoint-based plan state restoration in Epic #8493. The command restores a plan's sandbox to the state captured at a given checkpoint, discarding all decisions made after that checkpoint. The checkpoint can be specified as an optional positional second argument or via the `--to-checkpoint` named option. Supports `--yes/-y` flag to skip confirmation prompts and `--format/-f` for output format selection (rich/plain/json/yaml). Included with comprehensive BDD test coverage (>= 97%) and spec-aligned output formatting showing rollback summary, changes reverted, impact analysis, and post-rollback state panels.
 
 ### Fixed
+- **InvariantService now persists invariants to SQLite database (issue #8573):**
+  ``InvariantService`` was using pure in-memory storage, causing all
+  invariant constraints to be lost on CLI process restart. Added an
+  ``InvariantModel`` SQLAlchemy model for standalone invariant persistence,
+  a domain ``InvariantRepositoryProtocol`` and its SQLAlchemy-backed
+  ``InvariantRepository`` implementation, an Alembic migration creating
+  the ``invariants`` table (``m11_001_standalone_invariants.py``), and
+  updated ``InvariantService`` to accept an optional ``database_url``
+  parameter. The application container passes this URL from Settings, so
+  invariants added via ``agents invariant add/list/remove`` survive process
+  restarts — all changes follow ADR-007 (Repository Pattern) conventions.
+
+- **Guard cleanup_stale against execute/processing and execute/complete plans** (#11121):
+  ``_create_sandbox_for_plan()`` in ``src/cleveragents/cli/commands/plan.py`` now
+  skips ``GitWorktreeSandbox.cleanup_stale()`` when the plan is in
+  ``execute/processing`` (execution in progress) or ``execute/complete`` (execution
+  finished, awaiting apply) state.  Previously, re-invoking ``agents plan execute``
+  on a completed plan would silently destroy the ``cleveragents/plan-<id>`` git
+  worktree branch, causing ``plan apply`` to merge zero artifacts.  The guard
+  preserves the branch per spec (§sandbox.cleanup defaults to ``on_apply``).
+
+- **Global CLI options ``--data-dir``, ``--config-path``, and ``-v`` now work correctly**
+  (#6785): These spec-required flags were absent from ``main_callback()`` in
+  ``src/cleveragents/cli/main.py``, causing any invocation with these flags to crash
+  with ``NoSuchOption: No such option``. All three options are now implemented per
+  ADR-021 §Global CLI Flags and ADR-024 §Resolution Chain.  ``--data-dir`` sets
+  ``CLEVERAGENTS_DATA_DIR`` before any ``Settings``-reading code runs, ``--config-path``
+  sets ``CLEVERAGENTS_CONFIG_PATH`` for ``ConfigService`` to pick up, and ``-v``
+  (repeatable count) maps to the appropriate ``structlog`` log level.
+
+- **Add regression test for ActorRegistry.add() nested provider/model extraction** (#4321):
+  Added BDD regression test confirming that provider and model are correctly
+  extracted from nested `actors.<name>.config` blocks when `type` is only at
+  the nested level and `name` uses the `local/` namespace prefix. This
+  scenario was previously fixed in #4300; the test ensures the fix remains
+  in place and prevents future regressions.
+
 - **Actor configuration validation incorrectly requires top-level provider field** (#4300): 
    Actor configuration in V3 is now obtained from the nested configuration 
    parameter, according to the specification.
@@ -270,6 +321,16 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ### Fixed
 
+- **`invariant_enforced` decisions not propagated to child plans on subplan spawn** (#9131):
+  Fixed `SubplanService.spawn()` to propagate all `invariant_enforced` decisions from the
+  parent plan's decision tree to each child plan's decision tree. Previously, child plans
+  started Strategize with a completely empty invariant set, violating the spec requirement:
+  "recorded as `invariant_enforced` decisions that propagate to child plans." The fix adds
+  a `_propagate_invariant_decisions()` helper that re-records each parent
+  `invariant_enforced` decision on the child plan, including `non_overridable` global
+  invariants. BDD regression coverage added in
+  `features/tdd_invariant_propagation_subplan.feature`.
+
 - **fix(repositories): derive PlanResult.success from result_success column instead of error_message** (#7501):
   Fixed a critical bug in `PlanRepository._to_domain` where `PlanResult.success` was incorrectly
   derived from `error_message is None`. Because `error_message` is shared between the build phase

CONTRIBUTORS.md

@@ -5,6 +5,7 @@
 * HAL 9000 <hal9000@cleverthis.com>
 * Hamza Khyari <hamza.khyari@cleverthis.com>
 * Jeffrey Phillips Freeman <jeffrey.freeman@syncleus.com>
+* Jeffrey Phillips Freeman <the@jeffreyfreeman.me>
 * Luis Mendes <luis.p.mendes@gmail.com>
 * Rui Hu <rui.hu@cleverthis.com>
 
@@ -13,11 +14,13 @@
 Below are some of the specific details of various contributions.
 
 * Jeffrey Phillips Freeman has acted as Lead Developer, daily contributor, and Project Owner.
+* Jeffrey Phillips Freeman has contributed an implementation for the invariant propagation fix (PR #10881 / issue #9131): added `_propagate_invariant_decisions()` to `SubplanService` to propagate all `invariant_enforced` decisions from parent plans to child plan decision trees during subplan spawn, satisfying the specification requirement for invariant propagation across hierarchical plan execution.
 * Brent E. Edwards has contributed quality assurance, test coverage, and CI pipeline improvements.
 * HAL 9000 has contributed automated implementation, bug fixes, and feature development as part of the CleverAgents automation pool.
 * HAL 9000 has contributed concurrency safety improvements, including thread-safe context tier management (issue #7547) for parallel plan execution.
 * HAL 9000 has contributed the plan concurrency race-condition fix (#7989): wired `LockService` into the plan lifecycle, guarding `execute_plan()` and `apply_plan()` with plan-level advisory locks and unique per-invocation owner identities to prevent silent concurrent state corruption.
-<<<<<<< HEAD
+* Jeffrey Phillips Freeman has contributed the InvariantService database persistence fix (PR #11166 / issue #8573): implemented SQLAlchemy-backed ``InvariantRepository``, added ``InvariantModel`` to the database models layer, created Alembic migration for the standalone ``invariants`` table, updated ``InvariantService`` with optional ``database_url`` parameter for cross-invocation persistence, and wired it into the application container (ADR-007 compliant).
+
 * HAL 9000 has contributed the bug-hunt-pool-supervisor non-blocking tracking fix (#7875 / PR #7957): updated step 5 to be best-effort and added rule 9 to prevent the automation-tracking-manager call from blocking the main supervisor loop.
 * Jeffrey Phillips Freeman has contributed the complete AUTO-BUG-POOL to AUTO-BUG-SUP tracking prefix fix across agent-system-specification.md, automation-tracking.md documentation and agent-system-specification.md spec document, replaced with correct `AUTO-BUG-SUP` prefix used by the bug-hunt-pool-supervisor agent (#7875).
 * HAL 9000 has contributed the plugin entry point security hardening fix (#7476): enforced entry point allowlist validation before importing plugin modules to prevent malicious plugin loading.

features/actor_registry_spec_yaml.feature

@@ -21,6 +21,13 @@ Feature: ActorRegistry.add() accepts spec-compliant actor YAML formats
     Then the actor should be registered with provider "anthropic" and model "claude-3"
     And the registered actor should exist in the actor service
 
+  @tdd_issue @tdd_issue_4321 @tdd_issue_4300
+  Scenario: registry.add() with nested-only type extracts provider/model from nested config
+    When I add a YAML with type only in nested actors map and provider/model in nested config
+    Then the actor should be registered with provider "custom" and model "fpga-model"
+    And the registered actor name should be "local/fpga-strategist"
+    And the registered actor should exist in the actor service
+
   # ── actors: map with unsafe flag ───────────────────────────────────
 
   Scenario: registry.add() preserves unsafe flag from nested spec-compliant config

features/cli_global_options.feature

@@ -0,0 +1,128 @@
+Feature: CLI global options --data-dir, --config-path, and -v
+
+  Background:
+    Given global options test env is clean
+
+  # -----------------------------------------------------------------------
+  # --data-dir option
+  # -----------------------------------------------------------------------
+
+  @tdd_issue @tdd_issue_6785
+  Scenario: --data-dir option is accepted with an existing directory
+    Given a temp data dir is prepared
+    When I run the global options CLI with data-dir flag and "version"
+    Then the global options CLI should succeed
+
+  Scenario: --data-dir option overrides CLEVERAGENTS_DATA_DIR for the invocation
+    Given a temp data dir is prepared
+    When I run the global options CLI with data-dir flag and "version"
+    Then the CLEVERAGENTS_DATA_DIR env var should match the temp data dir
+
+  Scenario: --data-dir with a path that is an existing file fails with a clear error
+    Given a temp file path is prepared
+    When I run the global options CLI with data-dir as temp file and "version"
+    Then the global options CLI should fail
+    And the global options output should contain "--data-dir"
+
+  Scenario: --data-dir stores value in ctx.obj for subcommands
+    Given a temp data dir is prepared
+    When I run the global options CLI with data-dir flag and "version"
+    Then the CLEVERAGENTS_DATA_DIR env var should match the temp data dir
+
+  # -----------------------------------------------------------------------
+  # --config-path option
+  # -----------------------------------------------------------------------
+
+  @tdd_issue @tdd_issue_6785
+  Scenario: --config-path option is accepted with an existing file
+    Given a temp config file is prepared
+    When I run the global options CLI with config-path flag and "version"
+    Then the global options CLI should succeed
+
+  Scenario: --config-path overrides CLEVERAGENTS_CONFIG_PATH for the invocation
+    Given a temp config file is prepared
+    When I run the global options CLI with config-path flag and "version"
+    Then the CLEVERAGENTS_CONFIG_PATH env var should match the temp config file
+
+  Scenario: --config-path with a non-existent file fails with a clear error
+    When I run the global options CLI with invalid config-path and "version"
+    Then the global options CLI should fail
+    And the global options output should contain "--config-path"
+
+  Scenario: --config-path stores value in ctx.obj for subcommands
+    Given a temp config file is prepared
+    When I run the global options CLI with config-path flag and "version"
+    Then the CLEVERAGENTS_CONFIG_PATH env var should match the temp config file
+
+  # -----------------------------------------------------------------------
+  # Combining --data-dir and --config-path
+  # -----------------------------------------------------------------------
+
+  Scenario: --data-dir and --config-path can be combined
+    Given a temp data dir is prepared
+    And a temp config file is prepared
+    When I run the global options CLI with both path flags and "version"
+    Then the global options CLI should succeed
+    And the CLEVERAGENTS_DATA_DIR env var should match the temp data dir
+    And the CLEVERAGENTS_CONFIG_PATH env var should match the temp config file
+
+  # -----------------------------------------------------------------------
+  # -v verbosity option
+  # -----------------------------------------------------------------------
+
+  Scenario: No -v flag results in CRITICAL (silent) log level
+    When I call main_callback with verbosity 0
+    Then the global options log level should be "CRITICAL"
+
+  Scenario: Single -v flag results in ERROR log level
+    When I call main_callback with verbosity 1
+    Then the global options log level should be "ERROR"
+
+  Scenario: -vv results in WARNING log level
+    When I call main_callback with verbosity 2
+    Then the global options log level should be "WARNING"
+
+  Scenario: -vvv results in INFO log level
+    When I call main_callback with verbosity 3
+    Then the global options log level should be "INFO"
+
+  Scenario: -vvvv results in DEBUG log level
+    When I call main_callback with verbosity 4
+    Then the global options log level should be "DEBUG"
+
+  Scenario: -vvvvv or more results in DEBUG log level
+    When I call main_callback with verbosity 5
+    Then the global options log level should be "DEBUG"
+
+  @tdd_issue @tdd_issue_6785
+  Scenario: -v flag is accepted by the CLI without crashing
+    When I run the global options CLI with args "-v version"
+    Then the global options CLI should succeed
+
+  Scenario: -vvv flags are accepted by the CLI without crashing
+    When I run the global options CLI with args "-v -v -v version"
+    Then the global options CLI should succeed
+
+  # -----------------------------------------------------------------------
+  # --help shows all three options
+  # -----------------------------------------------------------------------
+
+  Scenario: --help output includes --data-dir option
+    When I run the global options CLI with args "--help"
+    Then the global options output should contain "--data-dir"
+
+  Scenario: --help output includes --config-path option
+    When I run the global options CLI with args "--help"
+    Then the global options output should contain "--config-path"
+
+  Scenario: -v flag appears in --help output
+    When I run the global options CLI with args "--help"
+    Then the global options output should contain "-v"
+
+  # -----------------------------------------------------------------------
+  # -v verbosity stored in ctx.obj
+  # -----------------------------------------------------------------------
+
+  Scenario: verbose count is stored in ctx.obj
+    When I call main_callback with verbosity 3
+    Then the global options ctx obj "verbose" should be 3

features/plan_generation_langgraph_coverage.feature

@@ -69,9 +69,20 @@ Feature: Plan Generation LangGraph Coverage
 
   Scenario: Should retry returns end when max retries reached
     Given I have a langgraph PlanGenerationGraph instance with max_retries 3
-    When I check langgraph should_retry with FAIL validation and retry_count 3
+    When I check langgraph should_retry with FAIL validation and retry_count 4
     Then the langgraph retry decision should be "end"
 
+  Scenario: Should retry still retries when retry_count equals max_retries
+    Given I have a langgraph PlanGenerationGraph instance with max_retries 3
+    When I check langgraph should_retry with FAIL validation and retry_count 3
+    Then the langgraph retry decision should be "retry"
+
+  Scenario: Validate node increments retry_count in returned state
+    Given I have a langgraph PlanGenerationGraph instance
+    When I execute the langgraph validate node with no changes
+    Then the langgraph validation status should be "FAIL"
+    And the langgraph validate result should include retry_count incremented from 0 to 1
+
   Scenario: Validate node fails when no changes provided
     Given I have a langgraph PlanGenerationGraph instance
     When I execute the langgraph validate node with no changes

features/plan_generation_uncovered_lines.feature

@@ -78,12 +78,13 @@ Feature: Plan Generation Graph Uncovered Lines Coverage
     And the langgraph async result should have generated_changes
     And the langgraph async result should have validation_result
 
-  # Testing line 403: Retry count increment in should_retry
-  Scenario: Should retry increments retry count correctly
+  # _should_retry is a read-only conditional-edge function; retry_count is
+  # incremented inside _validate (a node) so LangGraph persists the new value.
+  Scenario: Should retry does not mutate state and returns retry when retries remain
     Given I have a langgraph PlanGenerationGraph instance with max_retries 3
     And I have a langgraph state with retry_count 1
     When I check uncovered langgraph should_retry with FAIL validation and retry_count 1
-    Then the uncovered langgraph state retry_count should be incremented to 2
+    Then the uncovered langgraph state retry_count should not be mutated and remain 1
     And the langgraph retry decision should be "retry"
 
   # Testing full workflow with retry logic

features/steps/actor_registry_spec_yaml_steps.py

@@ -143,6 +143,23 @@ def step_add_actors_separate(context: Context) -> None:
     context.spec_result = context.spec_registry.add(yaml_text)
 
 
+@when(
+    "I add a YAML with type only in nested actors map and provider/model in nested config"
+)
+def step_add_nested_type_only(context: Context) -> None:
+    yaml_text = (
+        "name: local/fpga-strategist\n"
+        "description: An FPGA strategist agent\n"
+        "actors:\n"
+        "  fpga-strategist:\n"
+        "    type: llm\n"
+        "    config:\n"
+        "      provider: custom\n"
+        "      model: fpga-model\n"
+    )
+    context.spec_result = context.spec_registry.add(yaml_text)
+
+
 @when("I add a spec-compliant YAML with actors map and unsafe flag")
 def step_add_actors_unsafe(context: Context) -> None:
     yaml_text = (

features/steps/cli_global_options_steps.py

@@ -0,0 +1,346 @@
+"""Step definitions for CLI global options: --data-dir, --config-path, -v.
+
+Tests for issue #6785 — spec-required global options that were missing from
+main_callback() and caused fatal "No such option" crashes.
+
+All step patterns use the prefix "global options" to avoid any ambiguity
+with the many generic CLI step definitions in other step files.
+"""
+
+from __future__ import annotations
+
+import os
+import re
+import shutil
+import tempfile
+from pathlib import Path
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+from behave import given, then, when
+from behave.runner import Context
+from typer.testing import CliRunner
+
+from cleveragents.cli.formatting import OutputFormat
+from cleveragents.cli.main import app
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+# Regex to strip ANSI escape sequences (CSI sequences and OSC sequences).
+# Used to make output assertions environment-agnostic — Rich may emit ANSI
+# colour/style codes when the CI environment does not set TERM=dumb.
+_ANSI_ESCAPE_RE = re.compile(
+    r"\x1b(?:"
+    r"\[[0-9;]*[a-zA-Z]"  # CSI sequences (SGR, cursor, etc.)
+    r"|"
+    r"\][^\x07]*\x07"  # OSC sequences (hyperlink, title, etc.)
+    r")"
+)
+
+
+def _strip_ansi(text: str) -> str:
+    """Remove ANSI escape sequences from *text*."""
+    return _ANSI_ESCAPE_RE.sub("", text)
+
+
+def _invoke(args: list[str]) -> Any:
+    """Invoke the Typer app with the given args and return the result."""
+    runner = CliRunner()
+    return runner.invoke(app, args, catch_exceptions=True)
+
+
+def _combined_output(result: Any) -> str:
+    """Return combined output from a CliRunner result, with ANSI codes stripped.
+
+    In non-TTY environments (CI), Typer/Rich may split output between
+    stdout and stderr.  Check both explicitly so that ``--help`` output
+    assertions work regardless of the terminal detection.
+    """
+    stdout = _strip_ansi((getattr(result, "stdout", "") or "").strip())
+    stderr = _strip_ansi((getattr(result, "stderr", "") or "").strip())
+    combined = (
+        (stdout + "\n" + stderr).strip() if (stdout and stderr) else (stdout or stderr)
+    )
+    return combined or _strip_ansi((result.output or "").strip())
+
+
+# ---------------------------------------------------------------------------
+# Given steps
+# ---------------------------------------------------------------------------
+
+
+@given("global options test env is clean")
+def step_global_opts_env_clean(context: Context) -> None:
+    """Ensure test-specific env vars are not polluted from previous tests.
+
+    Registers a cleanup handler that restores the original env var values after
+    the scenario ends.  The global ``after_scenario`` hook in
+    ``features/environment.py`` runs all ``context._cleanup_handlers``.
+    """
+    orig_data_dir = os.environ.get("CLEVERAGENTS_DATA_DIR")
+    orig_config_path = os.environ.get("CLEVERAGENTS_CONFIG_PATH")
+    # Clean up any leftover from previous scenarios
+    for var in ("CLEVERAGENTS_DATA_DIR", "CLEVERAGENTS_CONFIG_PATH"):
+        os.environ.pop(var, None)
+
+    # Register restoration handler so that env vars set by main_callback()
+    # during CLI invocation do not pollute subsequent scenarios.
+    def _restore() -> None:
+        for key, val in [
+            ("CLEVERAGENTS_DATA_DIR", orig_data_dir),
+            ("CLEVERAGENTS_CONFIG_PATH", orig_config_path),
+        ]:
+            if val is not None:
+                os.environ[key] = val
+            else:
+                os.environ.pop(key, None)
+
+    if not hasattr(context, "_cleanup_handlers"):
+        context._cleanup_handlers = []
+    context._cleanup_handlers.append(_restore)
+
+
+@given("a temp data dir is prepared")
+def step_temp_data_dir_prepared(context: Context) -> None:
+    """Create a real temporary directory for --data-dir testing.
+
+    Registers a cleanup handler so the directory is removed after the scenario.
+    """
+    tmp_dir = tempfile.mkdtemp(prefix="ca_test_data_dir_")
+    context.temp_data_dir = tmp_dir
+    if not hasattr(context, "_cleanup_handlers"):
+        context._cleanup_handlers = []
+    context._cleanup_handlers.append(lambda: shutil.rmtree(tmp_dir, ignore_errors=True))
+
+
+@given("a temp config file is prepared")
+def step_temp_config_file_prepared(context: Context) -> None:
+    """Create a real temporary TOML config file for --config-path testing.
+
+    Registers a cleanup handler so the file is removed after the scenario.
+    """
+    with tempfile.NamedTemporaryFile(
+        suffix=".toml", prefix="ca_test_config_", delete=False
+    ) as tmp_file:
+        tmp_file.write(b"# test config\n")
+        tmp_path = tmp_file.name
+    context.temp_config_file = tmp_path
+    if not hasattr(context, "_cleanup_handlers"):
+        context._cleanup_handlers = []
+    context._cleanup_handlers.append(lambda: os.unlink(tmp_path))
+
+
+@given("a temp file path is prepared")
+def step_temp_file_path_prepared(context: Context) -> None:
+    """Create a temporary FILE (not directory) to trigger --data-dir validation.
+
+    Registers a cleanup handler so the file is removed after the scenario.
+    """
+    with tempfile.NamedTemporaryFile(prefix="ca_test_file_", delete=False) as tmp_file:
+        tmp_path = tmp_file.name
+    context.temp_file_path = tmp_path
+    if not hasattr(context, "_cleanup_handlers"):
+        context._cleanup_handlers = []
+    context._cleanup_handlers.append(lambda: os.unlink(tmp_path))
+
+
+# ---------------------------------------------------------------------------
+# When steps - CLI invocation
+# ---------------------------------------------------------------------------
+
+
+@when('I run the global options CLI with data-dir flag and "{subcommand}"')
+def step_run_global_opts_data_dir(context: Context, subcommand: str) -> None:
+    """Run the CLI with --data-dir pointing at the temp directory."""
+    args = ["--data-dir", context.temp_data_dir, subcommand]
+    result = _invoke(args)
+    context.global_opts_result = result
+    context.global_opts_exit_code = result.exit_code
+    context.global_opts_output = _combined_output(result)
+
+
+@when('I run the global options CLI with config-path flag and "{subcommand}"')
+def step_run_global_opts_config_path(context: Context, subcommand: str) -> None:
+    """Run the CLI with --config-path pointing at the temp config file."""
+    args = ["--config-path", context.temp_config_file, subcommand]
+    result = _invoke(args)
+    context.global_opts_result = result
+    context.global_opts_exit_code = result.exit_code
+    context.global_opts_output = _combined_output(result)
+
+
+@when('I run the global options CLI with both path flags and "{subcommand}"')
+def step_run_global_opts_both(context: Context, subcommand: str) -> None:
+    """Run the CLI with both --data-dir and --config-path."""
+    args = [
+        "--data-dir",
+        context.temp_data_dir,
+        "--config-path",
+        context.temp_config_file,
+        subcommand,
+    ]
+    result = _invoke(args)
+    context.global_opts_result = result
+    context.global_opts_exit_code = result.exit_code
+    context.global_opts_output = _combined_output(result)
+
+
+@when('I run the global options CLI with data-dir as temp file and "{subcommand}"')
+def step_run_global_opts_data_dir_file(context: Context, subcommand: str) -> None:
+    """Run CLI with --data-dir pointing to a FILE (triggers validation error)."""
+    args = ["--data-dir", context.temp_file_path, subcommand]
+    result = _invoke(args)
+    context.global_opts_result = result
+    context.global_opts_exit_code = result.exit_code
+    context.global_opts_output = _combined_output(result)
+
+
+@when('I run the global options CLI with invalid config-path and "{subcommand}"')
+def step_run_global_opts_invalid_config_path(context: Context, subcommand: str) -> None:
+    """Run CLI with --config-path pointing to a non-existent file."""
+    args = ["--config-path", "/tmp/no_such_file_xyz_99999.toml", subcommand]
+    result = _invoke(args)
+    context.global_opts_result = result
+    context.global_opts_exit_code = result.exit_code
+    context.global_opts_output = _combined_output(result)
+
+
+@when('I run the global options CLI with args "{args}"')
+def step_run_global_opts_with_args(context: Context, args: str) -> None:
+    """Run the CLI with space-separated args string."""
+    result = _invoke(args.split())
+    context.global_opts_result = result
+    context.global_opts_exit_code = result.exit_code
+    context.global_opts_output = _combined_output(result)
+
+
+# ---------------------------------------------------------------------------
+# When steps - direct callback testing for verbosity
+# ---------------------------------------------------------------------------
+
+
+@when("I call main_callback with verbosity {count:d}")
+def step_call_main_callback_verbosity(context: Context, count: int) -> None:
+    """Invoke main_callback directly, capturing the log level configured."""
+    from cleveragents.cli.main import main_callback
+
+    context.global_opts_captured_log_level: str | None = None
+    context.global_opts_captured_ctx_obj: dict[str, Any] = {}
+
+    def _capture_log_level(
+        *, env: str = "development", log_level: str = "INFO"
+    ) -> None:
+        context.global_opts_captured_log_level = log_level
+
+    mock_ctx = MagicMock()
+    mock_ctx.obj = {}
+
+    with patch(
+        "cleveragents.config.logging.configure_structlog",
+        side_effect=_capture_log_level,
+    ):
+        main_callback(
+            ctx=mock_ctx,
+            version=None,  # type: ignore[arg-type]
+            show_secrets=False,
+            fmt=OutputFormat.RICH,
+            data_dir=None,
+            config_path=None,
+            verbose=count,
+        )
+
+    context.global_opts_captured_ctx_obj = mock_ctx.obj
+
+
+# ---------------------------------------------------------------------------
+# Then steps - exit codes
+# ---------------------------------------------------------------------------
+
+
+@then("the global options CLI should succeed")
+def step_global_opts_cli_succeed(context: Context) -> None:
+    actual = context.global_opts_exit_code
+    output = context.global_opts_output
+    assert actual == 0, f"Expected exit code 0, got {actual}.\nOutput:\n{output}"
+
+
+@then("the global options CLI should fail")
+def step_global_opts_cli_fail(context: Context) -> None:
+    actual = context.global_opts_exit_code
+    assert actual != 0, (
+        f"Expected non-zero exit code, got {actual}.\nOutput:\n{context.global_opts_output}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Then steps - env var assertions
+# ---------------------------------------------------------------------------
+
+
+@then("the CLEVERAGENTS_DATA_DIR env var should match the temp data dir")
+def step_env_data_dir_matches(context: Context) -> None:
+    """Verify that --data-dir set CLEVERAGENTS_DATA_DIR to the temp directory."""
+    env_val = os.environ.get("CLEVERAGENTS_DATA_DIR")
+    expected = str(Path(context.temp_data_dir).resolve())
+    assert env_val is not None, (
+        "CLEVERAGENTS_DATA_DIR was not set after --data-dir invocation"
+    )
+    assert str(Path(env_val).resolve()) == expected, (
+        f"CLEVERAGENTS_DATA_DIR={env_val!r} does not match expected {expected!r}"
+    )
+
+
+@then("the CLEVERAGENTS_CONFIG_PATH env var should match the temp config file")
+def step_env_config_path_matches(context: Context) -> None:
+    """Verify that --config-path set CLEVERAGENTS_CONFIG_PATH to the temp file."""
+    env_val = os.environ.get("CLEVERAGENTS_CONFIG_PATH")
+    expected = str(Path(context.temp_config_file).resolve())
+    assert env_val is not None, (
+        "CLEVERAGENTS_CONFIG_PATH was not set after --config-path invocation"
+    )
+    assert str(Path(env_val).resolve()) == expected, (
+        f"CLEVERAGENTS_CONFIG_PATH={env_val!r} does not match expected {expected!r}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Then steps - ctx.obj assertions
+# ---------------------------------------------------------------------------
+
+
+@then('the global options ctx obj "{key}" should be {value:d}')
+def step_global_opts_ctx_obj_int(context: Context, key: str, value: int) -> None:
+    """Verify ctx.obj[key] has the expected integer value."""
+    obj = context.global_opts_captured_ctx_obj
+    assert key in obj, f"ctx.obj missing key '{key}'. ctx.obj={obj}"
+    actual = obj[key]
+    assert actual == value, f"ctx.obj['{key}']={actual!r}, expected {value!r}"
+
+
+# ---------------------------------------------------------------------------
+# Then steps - log level assertions
+# ---------------------------------------------------------------------------
+
+
+@then('the global options log level should be "{level}"')
+def step_global_opts_log_level(context: Context, level: str) -> None:
+    """Verify that configure_structlog was called with the expected log level."""
+    actual = context.global_opts_captured_log_level
+    assert actual is not None, "configure_structlog was not called"
+    assert actual.upper() == level.upper(), (
+        f"Expected log level {level!r}, got {actual!r}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Then steps - output assertions
+# ---------------------------------------------------------------------------
+
+
+@then('the global options output should contain "{text}"')
+def step_global_opts_output_contains(context: Context, text: str) -> None:
+    output = context.global_opts_output
+    assert text in output, f"Expected '{text}' in output:\n{output}"

features/steps/plan_generation_langgraph_coverage_steps.py

@@ -252,14 +252,18 @@ def step_langgraph_graph_with_max_retries(context: Any, retries: int) -> None:
     "I check langgraph should_retry with {status} validation and retry_count {count:d}"
 )
 def step_check_langgraph_should_retry(context: Any, status: str, count: int) -> None:
-    """Check should_retry decision."""
+    """Check should_retry decision.
+
+    Note: _should_retry is a conditional-edge function and is read-only from
+    LangGraph's perspective.  The state dict passed here is NOT mutated by
+    the function (retry_count is incremented inside _validate instead).
+    """
     state: dict[str, Any] = {
         "validation_result": {"status": status},
         "retry_count": count,
     }
     decision = context.graph._should_retry(state)
     context.retry_decision = decision
-    context.final_retry_count = state.get("retry_count", count)
 
 
 @then('the langgraph retry decision should be "{decision}"')
@@ -283,6 +287,28 @@ def step_langgraph_validation_status(context: Any, status: str) -> None:
     assert validation.get("status") == status
 
 
+@then(
+    "the langgraph validate result should include retry_count incremented from {before:d} to {after:d}"
+)
+def step_langgraph_validate_result_retry_count(
+    context: Any, before: int, after: int
+) -> None:
+    """Verify _validate returns an incremented retry_count.
+
+    The _validate node is responsible for incrementing retry_count so that
+    LangGraph persists the new value into the graph state.  Conditional-edge
+    functions such as _should_retry are read-only from LangGraph's perspective
+    and must not mutate state.
+    """
+    result = context.node_result
+    assert "retry_count" in result, (
+        "_validate must return retry_count in its result dict"
+    )
+    assert result["retry_count"] == after, (
+        f"Expected retry_count {after} (incremented from {before}), got {result['retry_count']}"
+    )
+
+
 @then('the langgraph validation message should contain "{text}"')
 def step_langgraph_validation_message_contains(context: Any, text: str) -> None:
     """Verify validation message contains text."""

features/steps/plan_generation_uncovered_lines_steps.py

@@ -459,16 +459,32 @@ def step_have_state_with_retry_count(context: Any, count: int) -> None:
     "I check uncovered langgraph should_retry with FAIL validation and retry_count {count:d}"
 )
 def step_check_should_retry_uncovered(context: Any, count: int) -> None:
-    """Check should_retry and verify retry_count increment."""
+    """Check should_retry returns correct decision.
+
+    _should_retry is a conditional-edge function — LangGraph treats it as
+    read-only.  The retry counter is already incremented inside _validate
+    (a node whose return dict IS merged into the state), so _handle_retry
+    is a simple pass-through bridge node.
+    """
     decision = context.graph._should_retry(context.state)
     context.retry_decision = decision
     context.final_retry_count = context.state.get("retry_count")
 
 
-@then("the uncovered langgraph state retry_count should be incremented to {expected:d}")
-def step_uncovered_retry_count_incremented(context: Any, expected: int) -> None:
-    """Verify retry_count was incremented."""
-    assert context.final_retry_count == expected
+@then(
+    "the uncovered langgraph state retry_count should not be mutated and remain {expected:d}"
+)
+def step_uncovered_retry_count_not_mutated(context: Any, expected: int) -> None:
+    """Verify _should_retry did not mutate retry_count in state.
+
+    _should_retry is a conditional-edge function; LangGraph treats it as
+    read-only.  The retry counter is incremented inside _validate (a node)
+    so that LangGraph persists the new value into the graph state.
+    """
+    assert context.final_retry_count == expected, (
+        f"_should_retry must not mutate state['retry_count']: "
+        f"expected {expected}, got {context.final_retry_count}"
+    )
 
 
 # Workflow retry scenario

features/steps/tdd_cleanup_stale_destroys_execute_output_steps.py

@@ -0,0 +1,318 @@
+"""Steps for tdd_cleanup_stale_destroys_execute_output.feature.
+
+TDD issue-capture test for bug #11121:
+_create_sandbox_for_plan() calls GitWorktreeSandbox.cleanup_stale() unconditionally,
+destroying the cleveragents/plan-<id> branch when the plan is already in
+execute/complete or execute/processing state (awaiting apply or still in progress).
+
+The scenarios are tagged @tdd_issue_11121 as permanent regression guards.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import shutil
+import subprocess
+import tempfile
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+from behave import given, then, when
+from behave.runner import Context
+
+_PLAN_ID = "01TDDSANDBOX000000000000A"
+_BRANCH_NAME = f"cleveragents/plan-{_PLAN_ID}"
+
+
+def _git(args: list[str], cwd: str) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(
+        ["git", *args],
+        cwd=cwd,
+        capture_output=True,
+        text=True,
+        check=True,
+        timeout=15,
+    )
+
+
+def _branch_exists(repo_path: str, branch: str) -> bool:
+    """Return True if the given branch exists in the repo."""
+    result = subprocess.run(
+        ["git", "rev-parse", "--verify", f"refs/heads/{branch}"],
+        cwd=repo_path,
+        capture_output=True,
+        text=True,
+        check=False,
+        timeout=10,
+    )
+    return result.returncode == 0
+
+
+def _init_git_repo(path: str) -> None:
+    _git(["init", "-q", "-b", "main"], path)
+    _git(["config", "user.name", "TDD Test"], path)
+    _git(["config", "user.email", "tdd@test.local"], path)
+    _git(["config", "commit.gpgsign", "false"], path)
+
+
+def _build_execute_complete_mocks(
+    context: Context,
+    repo_path: str,
+) -> None:
+    """Build mock service + container for a plan in execute/complete state."""
+    from cleveragents.domain.models.core.plan import PlanPhase, ProcessingState
+
+    mock_resource = MagicMock()
+    mock_resource.resource_type_name = "git-checkout"
+    mock_resource.location = repo_path
+    mock_resource.resource_id = "res-tdd-11121-test"
+
+    mock_lr = MagicMock()
+    mock_lr.resource_id = "res-tdd-11121-test"
+
+    mock_project = MagicMock()
+    mock_project.linked_resources = [mock_lr]
+
+    # Plan is in execute/complete state — this is the critical state for the bug
+    mock_plan = MagicMock()
+    mock_plan.project_links = [MagicMock(project_name="local/tdd-11121-project")]
+    mock_plan.phase = PlanPhase.EXECUTE
+    mock_plan.processing_state = ProcessingState.COMPLETE
+    mock_plan.state = ProcessingState.COMPLETE
+
+    mock_service = MagicMock()
+    mock_service.get_plan.return_value = mock_plan
+
+    mock_project_repo = MagicMock()
+    mock_project_repo.get.return_value = mock_project
+
+    mock_resource_registry = MagicMock()
+    mock_resource_registry.show_resource.return_value = mock_resource
+
+    mock_container = MagicMock()
+    mock_container.namespaced_project_repo.return_value = mock_project_repo
+    mock_container.resource_registry_service.return_value = mock_resource_registry
+
+    context.tdd11121_service = mock_service
+    context.tdd11121_container = mock_container
+    context.tdd11121_repo_path = repo_path
+    context.tdd11121_plan_id = _PLAN_ID
+    context.tdd11121_branch = _BRANCH_NAME
+
+
+@given("a temp git repo with an execute-output branch for tdd 11121")
+def step_create_git_repo_with_execute_output(context: Context) -> None:
+    """Create a real git repo with a cleveragents/plan-<id> branch holding execute output.
+
+    This simulates the state after a successful plan execute: the worktree branch
+    exists and contains at least one committed file representing execution output.
+    """
+    d = tempfile.mkdtemp(prefix="tdd-11121-")
+    context.add_cleanup(shutil.rmtree, d, True)
+
+    # Initialise the repo with a base commit on main
+    _init_git_repo(d)
+    Path(d, "README.md").write_text("# project\n")
+    _git(["add", "."], d)
+    _git(["commit", "-q", "-m", "init: base commit"], d)
+
+    # Create the cleveragents/plan-<id> branch and commit an output file to it.
+    # This simulates what _commit_worktree_changes() does after execute completes.
+    _git(["checkout", "-q", "-b", _BRANCH_NAME], d)
+    output_file = Path(d, "generated_output.py")
+    output_file.write_text("# Generated by plan execute\nresult = 42\n")
+    _git(["add", "."], d)
+    _git(["commit", "-q", "-m", f"cleveragents: execute output for plan {_PLAN_ID}"], d)
+
+    # Return to main so the repo is in a normal state
+    _git(["checkout", "-q", "main"], d)
+
+    context.tdd11121_repo_path = d
+    # Verify the branch exists before the test
+    assert _branch_exists(d, _BRANCH_NAME), (
+        f"Pre-condition failed: branch {_BRANCH_NAME} should exist before test"
+    )
+
+
+def _build_execute_processing_mocks(
+    context: Context,
+    repo_path: str,
+) -> None:
+    """Build mock service + container for a plan in execute/processing state."""
+    from cleveragents.domain.models.core.plan import PlanPhase, ProcessingState
+
+    mock_resource = MagicMock()
+    mock_resource.resource_type_name = "git-checkout"
+    mock_resource.location = repo_path
+    mock_resource.resource_id = "res-tdd-11121-test"
+
+    mock_lr = MagicMock()
+    mock_lr.resource_id = "res-tdd-11121-test"
+
+    mock_project = MagicMock()
+    mock_project.linked_resources = [mock_lr]
+
+    mock_plan = MagicMock()
+    mock_plan.project_links = [MagicMock(project_name="local/tdd-11121-project")]
+    mock_plan.phase = PlanPhase.EXECUTE
+    mock_plan.processing_state = ProcessingState.PROCESSING
+    mock_plan.state = ProcessingState.PROCESSING
+
+    mock_service = MagicMock()
+    mock_service.get_plan.return_value = mock_plan
+
+    mock_project_repo = MagicMock()
+    mock_project_repo.get.return_value = mock_project
+
+    mock_resource_registry = MagicMock()
+    mock_resource_registry.show_resource.return_value = mock_resource
+
+    mock_container = MagicMock()
+    mock_container.namespaced_project_repo.return_value = mock_project_repo
+    mock_container.resource_registry_service.return_value = mock_resource_registry
+
+    context.tdd11121_service = mock_service
+    context.tdd11121_container = mock_container
+    context.tdd11121_repo_path = repo_path
+    context.tdd11121_plan_id = _PLAN_ID
+    context.tdd11121_branch = _BRANCH_NAME
+
+
+@given("a mocked plan service with the plan in execute/complete state for tdd 11121")
+def step_mock_service_execute_complete(context: Context) -> None:
+    """Set up mock service returning a plan in execute/complete state."""
+    _build_execute_complete_mocks(context, context.tdd11121_repo_path)
+
+
+@given("a mocked plan service with the plan in execute/processing state for tdd 11121")
+def step_mock_service_execute_processing(context: Context) -> None:
+    """Set up mock service returning a plan in execute/processing state."""
+    _build_execute_processing_mocks(context, context.tdd11121_repo_path)
+
+
+@when(
+    "I call _create_sandbox_for_plan a second time on the execute/complete plan for tdd 11121"
+)
+def step_call_create_sandbox_second_time(context: Context) -> None:
+    """Call _create_sandbox_for_plan on a plan already in execute/complete state.
+
+    This simulates the user re-running 'agents plan execute <PLAN_ID>' after
+    execution has already completed. The bug causes cleanup_stale to run and
+    destroy the cleveragents/plan-<id> branch that holds the execute output.
+    """
+    from cleveragents.cli.commands.plan import _create_sandbox_for_plan
+
+    context.tdd11121_second_call_exception: Exception | None = None
+    try:
+        with (
+            patch(
+                "cleveragents.cli.commands.plan.get_container",
+                return_value=context.tdd11121_container,
+            ),
+            patch(
+                "cleveragents.application.container.get_container",
+                return_value=context.tdd11121_container,
+            ),
+        ):
+            _sandbox_root, sandbox_infos = _create_sandbox_for_plan(
+                context.tdd11121_plan_id,
+                context.tdd11121_service,
+            )
+            context.tdd11121_sandbox_infos = sandbox_infos
+            # Clean up any newly created sandbox to avoid resource leaks
+            for sinfo in sandbox_infos:
+                with contextlib.suppress(Exception):
+                    sinfo.sandbox_obj.cleanup()
+    except Exception as exc:
+        # Record but don't re-raise — we want to check branch state regardless
+        context.tdd11121_second_call_exception = exc
+
+
+@when(
+    "I call _create_sandbox_for_plan a second time on the execute/processing plan for tdd 11121"
+)
+def step_call_create_sandbox_second_time_processing(context: Context) -> None:
+    """Call _create_sandbox_for_plan on a plan already in execute/processing state.
+
+    This simulates the user re-running 'agents plan execute <PLAN_ID>' while the
+    plan is still in progress (execute/processing). The guard must protect the
+    sandbox branch from cleanup_stale in this state too.
+    """
+    from cleveragents.cli.commands.plan import _create_sandbox_for_plan
+
+    context.tdd11121_second_call_exception: Exception | None = None
+    try:
+        with (
+            patch(
+                "cleveragents.cli.commands.plan.get_container",
+                return_value=context.tdd11121_container,
+            ),
+            patch(
+                "cleveragents.application.container.get_container",
+                return_value=context.tdd11121_container,
+            ),
+        ):
+            _sandbox_root, sandbox_infos = _create_sandbox_for_plan(
+                context.tdd11121_plan_id,
+                context.tdd11121_service,
+            )
+            context.tdd11121_sandbox_infos = sandbox_infos
+            for sinfo in sandbox_infos:
+                with contextlib.suppress(Exception):
+                    sinfo.sandbox_obj.cleanup()
+    except Exception as exc:
+        context.tdd11121_second_call_exception = exc
+
+
+@then(
+    "the cleveragents plan branch should still exist after the second call for tdd 11121"
+)
+def step_branch_still_exists(context: Context) -> None:
+    """Assert the cleveragents/plan-<id> branch was NOT destroyed by cleanup_stale.
+
+    The guard in _create_sandbox_for_plan skips cleanup_stale when the plan is
+    already in execute/processing or execute/complete state, preserving the
+    sandbox branch so plan apply can merge it.
+    """
+    repo_path = context.tdd11121_repo_path
+    branch = context.tdd11121_branch
+    assert _branch_exists(repo_path, branch), (
+        f"Bug #11121: branch '{branch}' was destroyed by cleanup_stale during "
+        f"a second _create_sandbox_for_plan call. "
+        f"The branch must survive until plan apply merges it."
+    )
+
+
+@then("the apply sandbox changes should find at least one artifact for tdd 11121")
+def step_apply_finds_artifacts(context: Context) -> None:
+    """Assert that apply can find the execute output after a re-invoked execute.
+
+    The guard in _create_sandbox_for_plan preserves the sandbox branch, so the
+    git diff between HEAD and the plan branch finds at least one artifact.
+    """
+    repo_path = context.tdd11121_repo_path
+    branch = context.tdd11121_branch
+
+    if not _branch_exists(repo_path, branch):
+        raise AssertionError(
+            f"Bug #11121: branch '{branch}' was destroyed by cleanup_stale. "
+            f"plan apply would find 0 artifacts (empty changeset). "
+            f"The branch must persist until apply merges it."
+        )
+
+    result = subprocess.run(
+        ["git", "diff", "--stat", f"HEAD...{branch}"],
+        cwd=repo_path,
+        capture_output=True,
+        text=True,
+        check=False,
+        timeout=10,
+    )
+    stat_lines = (result.stdout or "").strip().splitlines()
+    artifact_count = max(0, len(stat_lines) - 1) if stat_lines else 0
+
+    assert artifact_count > 0, (
+        f"Bug #11121: plan apply would find {artifact_count} artifacts after "
+        f"re-invoked execute on execute/complete plan. Expected >= 1 artifact. "
+        f"diff --stat output: {result.stdout!r}"
+    )

features/steps/tdd_invariant_persistence_steps.py

@@ -1,20 +1,10 @@
-"""Step definitions for tdd_invariant_persistence.feature (bug #1022).
+"""Step definitions for tdd_invariant_persistence.feature (bug #1022, now fixed).
 
-TDD issue-capture tests verifying that ``InvariantService`` persists invariants
-across simulated CLI process restarts (separate service instances).
-
-Bug #1022: ``InvariantService`` stores invariants in an in-memory dict
-(``self._invariants``) with no database persistence layer.  Each CLI
-invocation spawns a fresh process with a new ``InvariantService()``
-instance, so all invariants are lost when the process exits.
-
-These steps exercise the current (buggy) behaviour by creating fresh
-``InvariantService`` instances to simulate separate process invocations.
-When the bug is fixed, the service will use a database repository and
-fresh instances backed by the same database will share state.
-
-The tests carry ``@tdd_expected_fail`` so CI passes while the bug is
-unfixed.  The tag will be removed when bug #1022 is fixed.
+Tests verify that ``InvariantService`` persists invariants across simulated
+CLI process restarts (separate service instances), confirming that Bug #8573
+/#1022 is resolved: the database-backed InvariantService stores data in SQLite,
+so separate CLI invocations share the same underlying ``cleveragents.db`` and
+cross-instance data visibility is confirmed.
 """
 
 from __future__ import annotations

features/steps/tdd_invariant_propagation_subplan_steps.py

@@ -0,0 +1,331 @@
+"""Step definitions for TDD Issue #9131 — invariant_enforced decisions not
+propagated to child plans on subplan spawn.
+
+These steps exercise ``SubplanService.spawn()`` and verify that it propagates
+``invariant_enforced`` decisions from the parent plan to each child plan's
+decision tree.
+
+On ``master`` (before the fix), ``spawn()`` creates child Plan objects but
+does NOT record ``invariant_enforced`` decisions on them.  The assertions in
+these steps will **fail** until the bug is fixed.
+"""
+
+from __future__ import annotations
+
+from behave import given, then, when
+from behave.runner import Context
+
+from cleveragents.application.services.decision_service import DecisionService
+from cleveragents.application.services.subplan_service import (
+    SpawnEntry,
+    SpawnResult,
+    SubplanService,
+)
+from cleveragents.domain.models.core.decision import (
+    ContextSnapshot,
+    Decision,
+    DecisionType,
+)
+from cleveragents.domain.models.core.plan import (
+    ExecutionMode,
+    NamespacedName,
+    Plan,
+    PlanIdentity,
+    SubplanConfig,
+)
+
+_PARENT_PLAN_ID: str = "01HGZ6FE0AQDYTR4BXVQZ6PN00"
+_ROOT_PLAN_ID: str = "01HGZ6FE0AQDYTR4BXVQZ6RF00"
+_SPAWN_DEC_ID: str = "01HGZ6FE0AQDYTR4BXVQZ6DA00"
+
+
+def _make_parent_plan() -> Plan:
+    """Create a minimal parent plan for testing."""
+    return Plan(
+        identity=PlanIdentity(
+            plan_id=_PARENT_PLAN_ID,
+            root_plan_id=_ROOT_PLAN_ID,
+        ),
+        namespaced_name=NamespacedName(namespace="local", name="parent-plan"),
+        description="Parent plan for invariant propagation test",
+        action_name="local/parent-action",
+    )
+
+
+def _make_spawn_decision(
+    plan_id: str = _PARENT_PLAN_ID,
+) -> Decision:
+    """Create a subplan_spawn Decision."""
+    return Decision(
+        decision_id=_SPAWN_DEC_ID,
+        plan_id=plan_id,
+        decision_type=DecisionType.SUBPLAN_SPAWN,
+        sequence_number=10,
+        question="Should we spawn a child plan?",
+        chosen_option="local/sub-action",
+        context_snapshot=ContextSnapshot(),
+    )
+
+
+def _make_spawn_entry(plan_id: str = _PARENT_PLAN_ID) -> SpawnEntry:
+    """Create a single spawn entry."""
+    return SpawnEntry(
+        decision=_make_spawn_decision(plan_id),
+        action_name="local/sub-action",
+        description="Child plan for invariant propagation test",
+    )
+
+
+# ---------------------------------------------------------------------------
+# Given steps
+# ---------------------------------------------------------------------------
+
+
+@given("a parent plan with invariant_enforced decisions recorded")
+def step_parent_with_invariant_decisions(context: Context) -> None:
+    """Set up a parent plan with one invariant_enforced decision."""
+    context.parent_plan = _make_parent_plan()
+    context.decision_service = DecisionService()
+
+    # Record one invariant_enforced decision on the parent plan
+    inv_dec = context.decision_service.record_decision(
+        plan_id=_PARENT_PLAN_ID,
+        decision_type=DecisionType.INVARIANT_ENFORCED,
+        question="Is this invariant enforced?",
+        chosen_option="invariant enforced: do not modify production data",
+    )
+    context.parent_invariant_decisions = [inv_dec]
+    context.expected_invariant_count = 1
+
+    context.subplan_service = SubplanService(
+        decision_service=context.decision_service,
+    )
+    context.subplan_config = SubplanConfig(
+        execution_mode=ExecutionMode.SEQUENTIAL,
+    )
+
+
+@given("a parent plan with {count:d} invariant_enforced decisions recorded")
+def step_parent_with_n_invariant_decisions(context: Context, count: int) -> None:
+    """Set up a parent plan with N invariant_enforced decisions."""
+    context.parent_plan = _make_parent_plan()
+    context.decision_service = DecisionService()
+
+    inv_decisions: list[Decision] = []
+    for i in range(count):
+        inv_dec = context.decision_service.record_decision(
+            plan_id=_PARENT_PLAN_ID,
+            decision_type=DecisionType.INVARIANT_ENFORCED,
+            question=f"Is invariant {i} enforced?",
+            chosen_option=f"invariant enforced: constraint {i}",
+        )
+        inv_decisions.append(inv_dec)
+
+    context.parent_invariant_decisions = inv_decisions
+    context.expected_invariant_count = count
+
+    context.subplan_service = SubplanService(
+        decision_service=context.decision_service,
+    )
+    context.subplan_config = SubplanConfig(
+        execution_mode=ExecutionMode.SEQUENTIAL,
+    )
+
+
+@given("a parent plan with a non_overridable invariant_enforced decision")
+def step_parent_with_non_overridable_invariant(context: Context) -> None:
+    """Set up a parent plan with a non_overridable invariant_enforced decision."""
+    context.parent_plan = _make_parent_plan()
+    context.decision_service = DecisionService()
+
+    # Record a non_overridable invariant decision
+    inv_dec = context.decision_service.record_decision(
+        plan_id=_PARENT_PLAN_ID,
+        decision_type=DecisionType.INVARIANT_ENFORCED,
+        question="Is this non-overridable global invariant enforced?",
+        chosen_option="non_overridable invariant enforced: never delete user data",
+        rationale="Global non-overridable invariant must propagate to all child plans",
+    )
+    context.parent_invariant_decisions = [inv_dec]
+    context.non_overridable_chosen_option = inv_dec.chosen_option
+    context.expected_invariant_count = 1
+
+    context.subplan_service = SubplanService(
+        decision_service=context.decision_service,
+    )
+    context.subplan_config = SubplanConfig(
+        execution_mode=ExecutionMode.SEQUENTIAL,
+    )
+
+
+@given("a parent plan with no invariant_enforced decisions")
+def step_parent_with_no_invariant_decisions(context: Context) -> None:
+    """Set up a parent plan with no invariant_enforced decisions."""
+    context.parent_plan = _make_parent_plan()
+    context.decision_service = DecisionService()
+    context.parent_invariant_decisions = []
+    context.expected_invariant_count = 0
+
+    context.subplan_service = SubplanService(
+        decision_service=context.decision_service,
+    )
+    context.subplan_config = SubplanConfig(
+        execution_mode=ExecutionMode.SEQUENTIAL,
+    )
+
+
+@given("a valid spawn entry for the parent plan")
+def step_valid_spawn_entry(context: Context) -> None:
+    """Create a single valid spawn entry."""
+    context.spawn_entries = [_make_spawn_entry()]
+
+
+# ---------------------------------------------------------------------------
+# When steps
+# ---------------------------------------------------------------------------
+
+
+@when("I spawn a child plan via SubplanService")
+def step_spawn_child_plan(context: Context) -> None:
+    """Call SubplanService.spawn() and capture the result."""
+    context.spawn_error = None
+    try:
+        context.spawn_result = context.subplan_service.spawn(
+            parent_plan=context.parent_plan,
+            config=context.subplan_config,
+            spawn_entries=context.spawn_entries,
+        )
+    except Exception as exc:
+        context.spawn_error = exc
+        context.spawn_result = None
+
+
+# ---------------------------------------------------------------------------
+# Then steps
+# ---------------------------------------------------------------------------
+
+
+@then(
+    "the decision service should have recorded invariant_enforced decisions"
+    " for the child plan"
+)
+def step_child_has_invariant_decisions(context: Context) -> None:
+    """Assert that invariant_enforced decisions were recorded for the child plan.
+
+    Bug #9131: spawn() does not propagate invariant_enforced decisions to
+    child plans.  This assertion will fail until the bug is fixed.
+    """
+    assert context.spawn_error is None, (
+        f"Spawn raised an unexpected error: {context.spawn_error}"
+    )
+    result: SpawnResult = context.spawn_result
+    assert result is not None, "Spawn result is None"
+    assert len(result.child_plans) > 0, "No child plans were created"
+
+    child_plan: Plan = result.child_plans[0]
+    child_plan_id: str = child_plan.identity.plan_id
+
+    # Query the decision service for invariant_enforced decisions on the child plan
+    child_invariant_decisions: list[Decision] = context.decision_service.list_by_type(
+        child_plan_id,
+        DecisionType.INVARIANT_ENFORCED,
+    )
+
+    assert len(child_invariant_decisions) > 0, (
+        f"Child plan {child_plan_id!r} has no invariant_enforced decisions. "
+        f"Bug #9131: SubplanService.spawn() does not propagate "
+        f"invariant_enforced decisions from parent plan to child plans."
+    )
+
+
+@then("the child plan should have {count:d} invariant_enforced decisions recorded")
+def step_child_has_n_invariant_decisions(context: Context, count: int) -> None:
+    """Assert that the child plan has exactly N invariant_enforced decisions.
+
+    Bug #9131: spawn() does not propagate invariant_enforced decisions.
+    """
+    assert context.spawn_error is None, (
+        f"Spawn raised an unexpected error: {context.spawn_error}"
+    )
+    result: SpawnResult = context.spawn_result
+    assert result is not None, "Spawn result is None"
+    assert len(result.child_plans) > 0, "No child plans were created"
+
+    child_plan: Plan = result.child_plans[0]
+    child_plan_id: str = child_plan.identity.plan_id
+
+    child_invariant_decisions: list[Decision] = context.decision_service.list_by_type(
+        child_plan_id,
+        DecisionType.INVARIANT_ENFORCED,
+    )
+
+    assert len(child_invariant_decisions) == count, (
+        f"Child plan {child_plan_id!r} has {len(child_invariant_decisions)} "
+        f"invariant_enforced decisions, expected {count}. "
+        f"Bug #9131: SubplanService.spawn() does not propagate "
+        f"invariant_enforced decisions from parent plan to child plans."
+    )
+
+
+@then("the child plan should have the non_overridable invariant decision propagated")
+def step_child_has_non_overridable_invariant(context: Context) -> None:
+    """Assert that the non_overridable invariant decision was propagated.
+
+    Bug #9131: spawn() does not propagate invariant_enforced decisions.
+    """
+    assert context.spawn_error is None, (
+        f"Spawn raised an unexpected error: {context.spawn_error}"
+    )
+    result: SpawnResult = context.spawn_result
+    assert result is not None, "Spawn result is None"
+    assert len(result.child_plans) > 0, "No child plans were created"
+
+    child_plan: Plan = result.child_plans[0]
+    child_plan_id: str = child_plan.identity.plan_id
+
+    child_invariant_decisions: list[Decision] = context.decision_service.list_by_type(
+        child_plan_id,
+        DecisionType.INVARIANT_ENFORCED,
+    )
+
+    assert len(child_invariant_decisions) > 0, (
+        f"Child plan {child_plan_id!r} has no invariant_enforced decisions. "
+        f"Bug #9131: non_overridable invariant was not propagated."
+    )
+
+    # Verify the non_overridable invariant text was propagated
+    expected_option: str = context.non_overridable_chosen_option
+    propagated_options: list[str] = [d.chosen_option for d in child_invariant_decisions]
+    assert expected_option in propagated_options, (
+        f"Non-overridable invariant chosen_option {expected_option!r} not found "
+        f"in child plan decisions: {propagated_options}. "
+        f"Bug #9131: non_overridable invariant was not propagated."
+    )
+
+
+@then("the spawn should succeed with no invariant decisions recorded for the child")
+def step_spawn_succeeds_no_invariants(context: Context) -> None:
+    """Assert that spawn succeeds cleanly when parent has no invariant decisions."""
+    assert context.spawn_error is None, (
+        f"Spawn raised an unexpected error: {context.spawn_error}"
+    )
+    result: SpawnResult = context.spawn_result
+    assert result is not None, "Spawn result is None"
+    assert result.total_spawned == 1, (
+        f"Expected 1 spawned child, got {result.total_spawned}"
+    )
+
+    # When parent has no invariant decisions, child should also have none
+    assert len(result.child_plans) > 0, "No child plans were created"
+    child_plan: Plan = result.child_plans[0]
+    child_plan_id: str = child_plan.identity.plan_id
+
+    child_invariant_decisions: list[Decision] = context.decision_service.list_by_type(
+        child_plan_id,
+        DecisionType.INVARIANT_ENFORCED,
+    )
+
+    assert len(child_invariant_decisions) == 0, (
+        f"Child plan {child_plan_id!r} has {len(child_invariant_decisions)} "
+        f"invariant_enforced decisions, expected 0 (parent had none)."
+    )

features/tdd_cleanup_stale_destroys_execute_output.feature

@@ -0,0 +1,33 @@
+@tdd_issue @tdd_issue_11121
+Feature: TDD Issue #11121 — cleanup_stale destroys git worktree branch on re-invoked execute
+  As a developer
+  I want to verify that _create_sandbox_for_plan does NOT delete the cleveragents/plan-<id>
+  branch when the plan is already in execute/complete state
+  So that plan apply can subsequently find and merge the correct artifacts
+
+  Bug #11121: _create_sandbox_for_plan() calls GitWorktreeSandbox.cleanup_stale()
+  unconditionally every time agents plan execute is invoked — including when the plan
+  is already in execute/complete state (execution finished, awaiting apply). This deletes
+  the cleveragents/plan-<id> git branch that holds the execution output, so when
+  agents plan apply runs next it finds no branch and merges zero artifacts.
+
+  @tdd_issue @tdd_issue_11121 @mock_only
+  Scenario: cleveragents/plan-<id> branch survives a second _create_sandbox_for_plan call when plan is execute/complete
+    Given a temp git repo with an execute-output branch for tdd 11121
+    And a mocked plan service with the plan in execute/complete state for tdd 11121
+    When I call _create_sandbox_for_plan a second time on the execute/complete plan for tdd 11121
+    Then the cleveragents plan branch should still exist after the second call for tdd 11121
+
+  @tdd_issue @tdd_issue_11121 @mock_only
+  Scenario: plan apply finds non-zero artifacts after re-invoked execute on execute/complete plan
+    Given a temp git repo with an execute-output branch for tdd 11121
+    And a mocked plan service with the plan in execute/complete state for tdd 11121
+    When I call _create_sandbox_for_plan a second time on the execute/complete plan for tdd 11121
+    Then the apply sandbox changes should find at least one artifact for tdd 11121
+
+  @tdd_issue @tdd_issue_11121 @mock_only
+  Scenario: cleveragents/plan-<id> branch survives a second _create_sandbox_for_plan call when plan is execute/processing
+    Given a temp git repo with an execute-output branch for tdd 11121
+    And a mocked plan service with the plan in execute/processing state for tdd 11121
+    When I call _create_sandbox_for_plan a second time on the execute/processing plan for tdd 11121
+    Then the cleveragents plan branch should still exist after the second call for tdd 11121

features/tdd_invariant_persistence.feature

@@ -1,49 +1,45 @@
-# TDD issue-capture test for bug #1022 — InvariantService in-memory storage only.
+# TDD issue-capture test for bug #1022 — InvariantService persistence.
 #
-# InvariantService stores invariants in an in-memory dict (self._invariants)
-# with no database persistence layer.  Each CLI invocation spawns a fresh
-# process with a new InvariantService() instance, so all invariants added in
-# one invocation are lost when the process exits.
+# Bug #1022 has been fixed: InvariantService now uses SQLite-based storage
+# via a lazy session-factory pattern.  Invariants added in one CLI invocation
+# persist across process restarts because all instances share the same
+# ``cleveragents.db`` (or equivalent) database configured in Settings.
 #
-# These scenarios prove the bug exists by simulating separate CLI invocations
-# (fresh InvariantService instances) and asserting that data added in one
-# invocation is visible in the next.  They FAIL until the bug is fixed.
-# The @tag inverts the result so CI passes.
-#
-# See: https://git.cleverthis.com/cleveragents/cleveragents-core/issues/1022
+# These scenarios verify cross-instance data visibility by simulating
+# separate CLI invocations (fresh service instances backing a shared DB).
 
-@tdd_issue @tdd_issue_1022 @mock_only
-Feature: TDD Issue #1022 — InvariantService invariants lost across process restarts
+@tdd_issue_1022 @mock_only
+Feature: TDD Issue #1022 — InvariantService persistence across process restarts
   As a developer using the agents CLI
   I want invariants added via "agents invariant add" to persist across CLI invocations
   So that "agents invariant list" in a subsequent invocation returns previously added invariants
 
-  InvariantService uses in-memory dict storage only.  Each CLI invocation
-  creates a fresh InvariantService() instance, so invariants are lost when
-  the process exits.  These tests simulate separate process invocations by
+  InvariantService uses SQLite-based storage backed by the configured database URL.
+  Fresh service instances share the same underlying database, so data persists across
+  process restarts.  These tests simulate separate process invocations by
   creating fresh service instances and verifying cross-instance data visibility.
 
-  @tdd_issue @tdd_issue_4283 @tdd_expected_fail
+  @tdd_issue_1022
   Scenario: Invariant added in one service instance is visible in a fresh instance
     Given I add a project invariant "All APIs must validate auth tokens" to project "local/api-service" via invariant service instance A
     When I create a fresh invariant service instance B
     And I list project invariants for "local/api-service" via instance B
     Then the invariant list from instance B should contain "All APIs must validate auth tokens"
 
-  @tdd_issue @tdd_issue_4283 @tdd_expected_fail
+  @tdd_issue_1022
   Scenario: Global invariant persists across simulated process restarts
     Given I add a global invariant "Never delete production data" via invariant service instance A
     When I create a fresh invariant service instance B
     And I list global invariants via instance B
     Then the invariant list from instance B should contain "Never delete production data"
 
-  @tdd_issue @tdd_issue_4283 @tdd_expected_fail
+  @tdd_issue_1022
   Scenario: Invariant added via CLI add is visible via CLI list in a new invocation
     Given I invoke invariant add via CLI with "--project local/webapp" and text "All changes need tests" using service invocation 1
     When I invoke invariant list via CLI with "--project local/webapp" using service invocation 2
     Then the CLI list output from invocation 2 should contain "All changes need tests"
 
-  @tdd_issue @tdd_issue_4283 @tdd_expected_fail
+  @tdd_issue_1022
   Scenario: Invariant soft-deleted in a fresh instance after being added in another
     Given I add a project invariant "Temporary constraint" to project "local/temp" via invariant service instance A
     And I capture the invariant ID from instance A

features/tdd_invariant_propagation_subplan.feature

@@ -0,0 +1,38 @@
+@tdd_issue @tdd_issue_9131 @mock_only
+Feature: TDD Issue #9131 — invariant_enforced decisions not propagated to child plans on subplan spawn
+  As a developer
+  I want to verify that SubplanService.spawn() propagates invariant_enforced decisions
+  from the parent plan to each child plan's decision tree
+  So that child plans start Strategize with the parent's enforced invariant set
+
+  The spec states: "recorded as invariant_enforced decisions that propagate to child plans."
+  SubplanService.spawn() currently creates child Plan objects with no invariant_enforced
+  decisions, violating the spec's propagation requirement.
+
+  @tdd_issue @tdd_issue_9131
+  Scenario: Child plan receives parent invariant_enforced decisions on spawn
+    Given a parent plan with invariant_enforced decisions recorded
+    And a valid spawn entry for the parent plan
+    When I spawn a child plan via SubplanService
+    Then the decision service should have recorded invariant_enforced decisions for the child plan
+
+  @tdd_issue @tdd_issue_9131
+  Scenario: Child plan receives all parent invariant_enforced decisions
+    Given a parent plan with 3 invariant_enforced decisions recorded
+    And a valid spawn entry for the parent plan
+    When I spawn a child plan via SubplanService
+    Then the child plan should have 3 invariant_enforced decisions recorded
+
+  @tdd_issue @tdd_issue_9131
+  Scenario: Non-overridable global invariants are propagated to child plans
+    Given a parent plan with a non_overridable invariant_enforced decision
+    And a valid spawn entry for the parent plan
+    When I spawn a child plan via SubplanService
+    Then the child plan should have the non_overridable invariant decision propagated
+
+  @tdd_issue @tdd_issue_9131
+  Scenario: Child plan with no parent invariant_enforced decisions spawns cleanly
+    Given a parent plan with no invariant_enforced decisions
+    And a valid spawn entry for the parent plan
+    When I spawn a child plan via SubplanService
+    Then the spawn should succeed with no invariant decisions recorded for the child

robot/cli_global_options.robot

@@ -0,0 +1,79 @@
+*** Settings ***
+Documentation    Integration tests for CLI global options --data-dir, --config-path, and -v.
+...
+...              Verifies that the spec-required global options are properly accepted
+...              and propagated end-to-end (issue #6785).
+Resource         ${CURDIR}/common.resource
+Suite Setup      Setup Test Environment
+Suite Teardown   Cleanup Test Environment
+
+*** Variables ***
+${HELPER}    ${CURDIR}/helper_cli_global_options.py
+
+*** Test Cases ***
+Data Dir Option Accepted With Valid Directory
+    [Documentation]    Verify --data-dir is accepted without error for an existing directory.
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    data-dir-accepted    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    cli-global-options-data-dir-accepted-ok
+
+Config Path Option Accepted With Valid File
+    [Documentation]    Verify --config-path is accepted without error for an existing file.
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    config-path-accepted    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    cli-global-options-config-path-accepted-ok
+
+Verbosity Flag Accepted
+    [Documentation]    Verify -v flag is accepted without crashing.
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    verbosity-accepted    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    cli-global-options-verbosity-accepted-ok
+
+Data Dir And Config Path Combined
+    [Documentation]    Verify --data-dir and --config-path can be combined end-to-end.
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    combined-options    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    cli-global-options-combined-ok
+
+Data Dir Overrides Settings Data Dir
+    [Documentation]    Verify --data-dir properly overrides CLEVERAGENTS_DATA_DIR.
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    data-dir-override    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    cli-global-options-data-dir-override-ok
+
+Config Path Overrides Config Service Path
+    [Documentation]    Verify --config-path properly overrides the config service path.
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    config-path-override    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    cli-global-options-config-path-override-ok
+
+Verbosity Sets Log Level Correctly
+    [Documentation]    Verify -v count correctly maps to log levels.
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    verbosity-levels    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    cli-global-options-verbosity-levels-ok
+
+Help Shows All Three Options
+    [Documentation]    Verify --help output includes --data-dir, --config-path, and -v.
+    ${result}=    Run Process    ${PYTHON}    -m    cleveragents    --help
+    ...    timeout=60s    on_timeout=kill
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    --data-dir
+    Should Contain    ${result.stdout}    --config-path
+    Should Contain    ${result.stdout}    -v

robot/helper_cli_global_options.py

@@ -0,0 +1,204 @@
+"""Helper script for cli_global_options.robot integration tests.
+
+Each sub-command verifies one end-to-end behaviour of the new global CLI
+options (--data-dir, --config-path, -v) introduced by issue #6785.
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+import tempfile
+from pathlib import Path
+from unittest.mock import patch
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _invoke(args: list[str]) -> tuple[int, str]:
+    """Invoke the Typer app and return (exit_code, combined_output)."""
+    from typer.testing import CliRunner
+
+    from cleveragents.cli.main import app
+
+    runner = CliRunner()
+    result = runner.invoke(app, args, catch_exceptions=True)
+    output = result.output or ""
+    return result.exit_code, output
+
+
+# ---------------------------------------------------------------------------
+# Test cases (sub-commands)
+# ---------------------------------------------------------------------------
+
+
+def test_data_dir_accepted() -> None:
+    with tempfile.TemporaryDirectory() as tmp:
+        code, output = _invoke(["--data-dir", tmp, "version"])
+        assert code == 0, f"Expected exit 0, got {code}. Output: {output}"
+    print("cli-global-options-data-dir-accepted-ok")
+
+
+def test_config_path_accepted() -> None:
+    with tempfile.NamedTemporaryFile(suffix=".toml", delete=False) as tmp:
+        tmp.write(b"# test config\n")
+        tmp_path = tmp.name
+    try:
+        code, output = _invoke(["--config-path", tmp_path, "version"])
+        assert code == 0, f"Expected exit 0, got {code}. Output: {output}"
+        print("cli-global-options-config-path-accepted-ok")
+    finally:
+        os.unlink(tmp_path)
+
+
+def test_verbosity_accepted() -> None:
+    code, output = _invoke(["-v", "version"])
+    assert code == 0, f"Expected exit 0, got {code}. Output: {output}"
+    print("cli-global-options-verbosity-accepted-ok")
+
+
+def test_combined_options() -> None:
+    with tempfile.TemporaryDirectory() as tmp_dir:
+        with tempfile.NamedTemporaryFile(suffix=".toml", delete=False) as tmp_cfg:
+            tmp_cfg.write(b"# test\n")
+            tmp_cfg_path = tmp_cfg.name
+        try:
+            code, output = _invoke(
+                [
+                    "--data-dir",
+                    tmp_dir,
+                    "--config-path",
+                    tmp_cfg_path,
+                    "version",
+                ]
+            )
+            assert code == 0, f"Expected exit 0, got {code}. Output: {output}"
+            print("cli-global-options-combined-ok")
+        finally:
+            os.unlink(tmp_cfg_path)
+
+
+def test_data_dir_override() -> None:
+    """Verify --data-dir sets CLEVERAGENTS_DATA_DIR env var during invocation."""
+    with tempfile.TemporaryDirectory() as tmp:
+        code, output = _invoke(["--data-dir", tmp, "version"])
+        assert code == 0, f"Expected exit 0, got {code}. Output: {output}"
+
+        # Verify the env var was set to (or derived from) the tmp dir
+        env_val = os.environ.get("CLEVERAGENTS_DATA_DIR")
+        if env_val:
+            resolved_env = str(Path(env_val).resolve())
+            resolved_tmp = str(Path(tmp).resolve())
+            assert resolved_env == resolved_tmp, (
+                f"CLEVERAGENTS_DATA_DIR={env_val!r} != {tmp!r}"
+            )
+
+    print("cli-global-options-data-dir-override-ok")
+
+
+def test_config_path_override() -> None:
+    """Verify --config-path sets CLEVERAGENTS_CONFIG_PATH env var during invocation."""
+    with tempfile.NamedTemporaryFile(suffix=".toml", delete=False) as tmp:
+        tmp.write(b"# test\n")
+        tmp_path = tmp.name
+    try:
+        code, output = _invoke(["--config-path", tmp_path, "version"])
+        assert code == 0, f"Expected exit 0, got {code}. Output: {output}"
+        env_val = os.environ.get("CLEVERAGENTS_CONFIG_PATH")
+        if env_val:
+            assert str(Path(env_val).resolve()) == str(Path(tmp_path).resolve()), (
+                f"CLEVERAGENTS_CONFIG_PATH={env_val!r} != {tmp_path!r}"
+            )
+        print("cli-global-options-config-path-override-ok")
+    finally:
+        os.unlink(tmp_path)
+
+
+def test_verbosity_levels() -> None:
+    """Verify the verbosity count → log level mapping."""
+    from unittest.mock import MagicMock
+
+    from cleveragents.cli.formatting import OutputFormat
+    from cleveragents.cli.main import main_callback
+
+    expected_levels = {
+        0: "CRITICAL",
+        1: "ERROR",
+        2: "WARNING",
+        3: "INFO",
+        4: "DEBUG",
+        5: "DEBUG",
+    }
+
+    for count, expected in expected_levels.items():
+        captured_levels: list[str] = []
+
+        def _make_capture(levels: list[str]):  # type: ignore[no-untyped-def]
+            def _capture(*, env: str = "development", log_level: str = "INFO") -> None:
+                levels.append(log_level)
+
+            return _capture
+
+        mock_ctx = MagicMock()
+        mock_ctx.obj = {}
+
+        with patch(
+            "cleveragents.config.logging.configure_structlog",
+            side_effect=_make_capture(captured_levels),
+        ):
+            main_callback(
+                ctx=mock_ctx,
+                version=None,  # type: ignore[arg-type]
+                show_secrets=False,
+                fmt=OutputFormat.RICH,
+                data_dir=None,
+                config_path=None,
+                verbose=count,
+            )
+
+        assert captured_levels, f"configure_structlog not called for count={count}"
+        actual = captured_levels[-1].upper()
+        assert actual == expected, (
+            f"verbose={count}: expected log_level={expected!r}, got {actual!r}"
+        )
+
+    print("cli-global-options-verbosity-levels-ok")
+
+
+# ---------------------------------------------------------------------------
+# Entry point
+# ---------------------------------------------------------------------------
+
+
+_COMMANDS: dict[str, object] = {
+    "data-dir-accepted": test_data_dir_accepted,
+    "config-path-accepted": test_config_path_accepted,
+    "verbosity-accepted": test_verbosity_accepted,
+    "combined-options": test_combined_options,
+    "data-dir-override": test_data_dir_override,
+    "config-path-override": test_config_path_override,
+    "verbosity-levels": test_verbosity_levels,
+}
+
+
+if __name__ == "__main__":
+    if len(sys.argv) < 2:
+        print("Usage: helper_cli_global_options.py <command>", file=sys.stderr)
+        sys.exit(1)
+
+    cmd = sys.argv[1]
+    fn = _COMMANDS.get(cmd)
+    if fn is None:
+        print(f"Unknown command: {cmd}", file=sys.stderr)
+        sys.exit(1)
+
+    try:
+        fn()  # type: ignore[call-arg]
+    except Exception as exc:
+        print(f"FAILED: {exc}", file=sys.stderr)
+        import traceback
+
+        traceback.print_exc(file=sys.stderr)
+        sys.exit(1)

robot/helper_tdd_invariant_persistence.py

@@ -1,12 +1,13 @@
-"""Helper script for tdd_invariant_persistence.robot (bug #1022).
+"""Helper script for tdd_invariant_persistence.robot (bug #1022, now fixed).
 
 Exercises InvariantService cross-invocation persistence at the integration
 level.  Each subcommand simulates a fresh CLI process by creating a new
 InvariantService instance, mirroring how the real CLI works (each
 ``python -m cleveragents`` call gets its own service).
 
-Bug #1022: InvariantService stores invariants in an in-memory dict only.
-Invariants added in one CLI invocation are lost when the process exits.
+Bug #8573 / #1022 is now FIXED: InvariantService uses SQLite-based persistence
+via the configured ``database_url``, so invariants added in one CLI invocation
+persist across process restarts.
 
 This helper is called from Robot Framework via ``Run Process``.
 """

robot/plan_generation_graph.robot

@@ -83,9 +83,10 @@ Plan Generation Graph Builds Workflow With Correct Nodes
     ...    assert 'analyze_requirements' in nodes
     ...    assert 'generate_plan' in nodes
     ...    assert 'validate' in nodes
+    ...    assert 'handle_retry' in nodes
     ...    print(f'Graph has {len(nodes)} nodes')
     ${result}=    Run Process    ${PYTHON}    -c    ${script}    shell=True
-    Should Contain    ${result.stdout}    Graph has 4 nodes
+    Should Contain    ${result.stdout}    Graph has 5 nodes
     Should Be Equal As Integers    ${result.rc}    0
 
 LangGraph Graphs Package Exports Workflow Classes
@@ -192,7 +193,9 @@ Load Context Node Generates Summary With Sample Contexts
 
 
 Should Retry Returns Retry When Validation Fails And Retries Available
-    [Documentation]    Test _should_retry returns "retry" appropriately
+    [Documentation]    Test _should_retry returns "retry" appropriately.
+    ...    Note: _should_retry is a read-only conditional-edge function in LangGraph.
+    ...    It does NOT mutate state; retry_count is incremented inside _validate.
     ${script}=    Catenate    SEPARATOR=\n
     ...    import sys
     ...    sys.path.insert(0, '${SRC_DIR}')
@@ -205,7 +208,7 @@ Should Retry Returns Retry When Validation Fails And Retries Available
     ...    }
     ...    decision = graph._should_retry(state)
     ...    assert decision == 'retry'
-    ...    assert state['retry_count'] == 1
+    ...    assert state['retry_count'] == 0, f'_should_retry must not mutate state, got {state["retry_count"]}'
     ...    print('Should retry: retry decision correct')
     ${result}=    Run Process    ${PYTHON}    -c    ${script}    shell=True
     Should Contain    ${result.stdout}    Should retry: retry decision correct
@@ -231,7 +234,9 @@ Should Retry Returns End When Validation Passes
     Should Be Equal As Integers    ${result.rc}    0
 
 Should Retry Returns End When Max Retries Reached
-    [Documentation]    Test _should_retry returns "end" when max retries reached
+    [Documentation]    Test _should_retry returns "end" when max retries exceeded.
+    ...    Because _validate increments retry_count before _should_retry is called,
+    ...    the boundary is retry_count > max_retries (i.e. retry_count=4 for max_retries=3).
     ${script}=    Catenate    SEPARATOR=\n
     ...    import sys
     ...    sys.path.insert(0, '${SRC_DIR}')
@@ -240,7 +245,7 @@ Should Retry Returns End When Max Retries Reached
     ...    graph = PlanGenerationGraph(llm=FakeListLLM(responses=['test']*3), max_retries=3)
     ...    state = {
     ...        'validation_result': {'status': 'FAIL'},
-    ...        'retry_count': 3
+    ...        'retry_count': 4
     ...    }
     ...    decision = graph._should_retry(state)
     ...    assert decision == 'end'

robot/tdd_invariant_persistence.robot

@@ -1,12 +1,8 @@
 *** Settings ***
-Documentation    TDD Issue #1022 — InvariantService invariants lost across CLI invocations
-...              Integration tests verifying that invariants added via the CLI
-...              persist across simulated process restarts.  InvariantService
-...              stores invariants in an in-memory dict only, so each CLI
-...              invocation starts with an empty service.  These tests exercise
-...              add-then-list and add-then-remove across fresh service
-...              instances.  They fail until the bug is fixed; the
-...              tag inverts the result so CI passes.
+Documentation    TDD Issue #1022 — InvariantService invariant persistence
+...              These tests verify that invariants added via the CLI persist
+...              across simulated process restarts.  Each test is a separate
+...              CLI invocation that adds then lists or removes an invariant.
 Resource         ${CURDIR}/common.resource
 Suite Setup      Setup Test Environment
 Suite Teardown   Cleanup Test Environment
@@ -17,7 +13,7 @@ ${HELPER}    ${CURDIR}/helper_tdd_invariant_persistence.py
 *** Test Cases ***
 TDD Invariant Add Then List Project Across Invocations
     [Documentation]    Add a project invariant in invocation 1, list in invocation 2.
-    [Tags]  tdd_issue    tdd_issue_1022    tdd_issue    tdd_issue_4318    tdd_expected_fail
+    [Tags]  tdd_issue    tdd_issue_1022    tdd_issue    tdd_issue_4318
 
     ${result}=    Run Process    ${PYTHON}    ${HELPER}    add-then-list-project    cwd=${WORKSPACE}    timeout=30s    on_timeout=kill
     Log    ${result.stdout}
@@ -27,7 +23,7 @@ TDD Invariant Add Then List Project Across Invocations
 
 TDD Invariant Add Then List Global Across Invocations
     [Documentation]    Add a global invariant in invocation 1, list in invocation 2.
-    [Tags]  tdd_issue    tdd_issue_1022    tdd_issue    tdd_issue_4318    tdd_expected_fail
+    [Tags]  tdd_issue    tdd_issue_1022    tdd_issue    tdd_issue_4318
 
     ${result}=    Run Process    ${PYTHON}    ${HELPER}    add-then-list-global    cwd=${WORKSPACE}    timeout=30s    on_timeout=kill
     Log    ${result.stdout}
@@ -37,7 +33,7 @@ TDD Invariant Add Then List Global Across Invocations
 
 TDD Invariant Remove Cross Instance
     [Documentation]    Add invariant in instance 1, remove by ID in instance 2.
-    [Tags]  tdd_issue    tdd_issue_1022    tdd_issue    tdd_issue_4318    tdd_expected_fail
+    [Tags]  tdd_issue    tdd_issue_1022    tdd_issue    tdd_issue_4318
 
     ${result}=    Run Process    ${PYTHON}    ${HELPER}    add-then-remove-cross    cwd=${WORKSPACE}    timeout=30s    on_timeout=kill
     Log    ${result.stdout}

src/cleveragents/agents/graphs/plan_generation.py

@@ -154,6 +154,8 @@ class PlanGenerationGraph:
     2. analyze_requirements: Analyzes user prompt for requirements
     3. generate_plan: Generates code changes based on requirements
     4. validate: Validates generated changes
+    5. handle_retry: Increments the retry counter (bridges conditional edge to
+       state update, since LangGraph conditional edges cannot persist mutations)
 
     The workflow includes conditional edges for retry logic and checkpointing
     for resumable execution.
@@ -273,6 +275,7 @@ class PlanGenerationGraph:
         workflow.add_node("analyze_requirements", self._analyze_requirements)
         workflow.add_node("generate_plan", self._generate_plan)
         workflow.add_node("validate", self._validate)
+        workflow.add_node("handle_retry", self._handle_retry)
 
         # Set entry point
         workflow.set_entry_point("load_context")
@@ -282,12 +285,16 @@ class PlanGenerationGraph:
         workflow.add_edge("analyze_requirements", "generate_plan")
         workflow.add_edge("generate_plan", "validate")
 
+        # Route retries through handle_retry node to persist the retry_count
+        # increment (conditional edge functions cannot mutate state).
+        workflow.add_edge("handle_retry", "analyze_requirements")
+
         # Add conditional edge for retry logic
         workflow.add_conditional_edges(
             "validate",
             self._should_retry,
             {
-                "retry": "analyze_requirements",
+                "retry": "handle_retry",
                 "end": END,
             },
         )
@@ -497,9 +504,15 @@ class PlanGenerationGraph:
             state: Current workflow state
 
         Returns:
-            Updated state with validation results
+            Updated state with validation results and incremented retry_count.
+            The retry_count is incremented here (inside a node) so that
+            LangGraph persists the new value into the graph state.  It must
+            NOT be mutated inside ``_should_retry`` because conditional-edge
+            functions are read-only from LangGraph's perspective — any
+            mutations they make to the state dict are silently discarded.
         """
         changes = state.get("generated_changes", [])
+        retry_count = state.get("retry_count", 0)
 
         if not changes:
             return {
@@ -507,6 +520,7 @@ class PlanGenerationGraph:
                     "status": "FAIL",
                     "message": "No changes to validate",
                 },
+                "retry_count": retry_count + 1,
             }
 
         # Create validation chain
@@ -527,14 +541,15 @@ class PlanGenerationGraph:
             )
             validation = str(result)
 
-            # Simple validation check (in real implementation, parse the LLM response)
-            is_valid = "PASS" in validation.upper() or len(all_code) > 10
+            # Reliance on the LLM response to determine pass/fail.
+            is_valid = "PASS" in validation.upper()
 
             return {
                 "validation_result": {
                     "status": "PASS" if is_valid else "FAIL",
                     "message": validation,
                 },
+                "retry_count": retry_count + 1,
             }
 
         except Exception as e:
@@ -543,27 +558,53 @@ class PlanGenerationGraph:
                     "status": "FAIL",
                     "message": f"Validation failed: {e!s}",
                 },
+                "retry_count": retry_count + 1,
             }
 
     def _should_retry(self, state: PlanGenerationState) -> str:
         """Determine if workflow should retry based on validation.
 
+        This is a conditional-edge function — LangGraph treats it as read-only.
+        Any mutations made to *state* here are silently discarded and never
+        persisted back into the graph state.  The retry counter is therefore
+        incremented inside ``_validate`` (a proper node whose return dict IS
+        merged into the state) so that the counter advances correctly across
+        iterations.
+
+        Args:
+            state: Current workflow state (read-only in this context)
+
+        Returns:
+            "retry" or "end" based on validation status and retry count
+        """
+        validation = state.get("validation_result", {})
+        # retry_count was already incremented by _validate before this edge
+        # function is called, so compare against max_retries directly.
+        retry_count = state.get("retry_count", 0)
+
+        # Check if validation failed and retries available
+        if validation.get("status") == "FAIL" and retry_count <= self.max_retries:
+            return "retry"
+
+        return "end"
+
+    def _handle_retry(self, state: PlanGenerationState) -> dict[str, Any]:
+        """Bridge node for the retry conditional edge.
+
+        Conditional edge functions in LangGraph cannot persist state
+        mutations — this node exists solely to satisfy the graph
+        compiler requirement that the retry path goes through a node
+        (not directly from a conditional edge back to a node).
+        The retry counter is already incremented inside ``_validate``
+        (a node whose return dict IS merged into the state).
+
         Args:
             state: Current workflow state
 
         Returns:
-            "retry" or "end" based on validation and retry count
+            Empty update (no state mutation needed)
         """
-        validation = state.get("validation_result", {})
-        retry_count = state.get("retry_count", 0)
-
-        # Check if validation failed and retries available
-        if validation.get("status") == "FAIL" and retry_count < self.max_retries:
-            # Increment retry count
-            state["retry_count"] = retry_count + 1
-            return "retry"
-
-        return "end"
+        return {}
 
     def _format_context_summary(self, contexts: list[Context]) -> str:
         """Format context files into a summary string.

src/cleveragents/application/container.py

@@ -104,6 +104,7 @@ from cleveragents.infrastructure.database.llm_trace_repository import (
 )
 from cleveragents.infrastructure.database.repositories import (
     CheckpointRepository,
+    GlobalInvariantRepository,
     NamespacedProjectRepository,
     ProjectResourceLinkRepository,
     SessionMessageRepository,
@@ -541,6 +542,27 @@ def _build_automation_profile_service(
     return AutomationProfileService(repo=profile_repo)
 
 
+def _build_global_invariant_repository(
+    database_url: str,
+) -> GlobalInvariantRepository:
+    """Build a ``GlobalInvariantRepository`` backed by the database URL.
+
+    Follows the same ``_build_*`` pattern used by ``_build_session_factory``,
+    ``_build_skill_service``, and other DB-backed helpers.  The repository is
+    constructed with ``auto_commit=True`` so that CLI commands outside a
+    ``UnitOfWork`` can commit inline immediately (Bug #8573).
+    """
+    from sqlalchemy import create_engine
+    from sqlalchemy.orm import sessionmaker
+
+    engine = create_engine(database_url, echo=False)
+    factory = sessionmaker(bind=engine, expire_on_commit=False)
+    return GlobalInvariantRepository(
+        session_factory=factory,
+        auto_commit=True,
+    )
+
+
 def _build_tool_registry_service(
     database_url: str,
 ) -> ToolRegistryService:
@@ -726,9 +748,17 @@ class Container(containers.DeclarativeContainer):
         event_bus=event_bus,
     )
 
-    # Invariant Service - Singleton (in-memory invariant management)
+    # Global Invariant Repository - Singleton (Bug #8573: persistence for CLI global/project invariants)
+    invariant_repo = providers.Singleton(
+        _build_global_invariant_repository,
+        database_url=database_url,
+    )
+
+    # Invariant Service - Singleton (delegates global/project scope CRUD to ``invariant_repo`` when wired)
     invariant_service = providers.Singleton(
         InvariantService,
+        database_url=database_url,
+        global_invariant_repo=invariant_repo,
     )
 
     # Lock Service - Singleton (shared advisory-lock state per process, #7989)

src/cleveragents/application/services/config_service.py

@@ -1183,6 +1183,14 @@ class ConfigService:
         project_root: Path | _AutoDiscover | None = _AUTO_DISCOVER,
     ) -> None:
         self._config_dir: Path = config_dir or _DEFAULT_CONFIG_DIR
+        # Honour CLEVERAGENTS_CONFIG_PATH env var when no explicit path is given.
+        # This allows the --config-path CLI flag (which sets the env var in
+        # main_callback) to propagate into all ConfigService instances created
+        # during subcommand execution (ADR-024 §Resolution Chain).
+        if config_path is None:
+            _env_config_path = os.environ.get("CLEVERAGENTS_CONFIG_PATH", "").strip()
+            if _env_config_path:
+                config_path = Path(_env_config_path)
         self._config_path: Path = config_path or (self._config_dir / "config.toml")
         self._event_bus = event_bus
         # ``_AUTO_DISCOVER`` means "auto-discover"; ``None`` means "no project root".

src/cleveragents/application/services/invariant_service.py

@@ -6,8 +6,15 @@ lifecycle operations.
 
 ## Storage
 
-Uses in-memory storage (same pattern as ``PlanLifecycleService``) with
-a dict keyed by invariant ID.
+Uses SQLite-based persistence via a lazy session-factory pattern (ADR-007).
+When a ``database_url`` is provided at construction, invariants are stored
+in the ``invariants`` table and persist across CLI invocations (process
+restarts).  When no ``database_url`` is provided the service falls back to
+pure in-memory mode (unchanged legacy behaviour — useful for testing).
+
+Standalone invariants are stored in the ``invariants`` table, separate
+from action-level (``action_invariants``) and plan-level
+(``plan_invariants``) child tables.
 
 ## Merge Precedence
 
@@ -22,6 +29,7 @@ from __future__ import annotations
 from typing import TYPE_CHECKING
 
 import structlog
+from sqlalchemy import create_engine as _create_engine
 from ulid import ULID
 
 from cleveragents.application.services.prompt_sanitizer import PromptSanitizer
@@ -36,6 +44,8 @@ from cleveragents.infrastructure.events.models import DomainEvent
 from cleveragents.infrastructure.events.types import EventType
 
 if TYPE_CHECKING:
+    from sqlalchemy.orm import Session, sessionmaker
+
     from cleveragents.infrastructure.events.protocol import EventBus
 
 logger = structlog.get_logger(__name__)
@@ -45,14 +55,29 @@ class InvariantService:
     """Service for managing invariant constraints.
 
     Provides add, list, remove (soft-delete), effective-set computation,
-    and enforcement record creation.  All storage is in-memory.
+    and enforcement record creation.  Storage is database-backed when a
+    ``database_url`` is provided at construction; otherwise in-memory.
     """
 
-    def __init__(self, event_bus: EventBus | None = None) -> None:
-        """Initialise the invariant service with empty in-memory storage.
+    def __init__(
+        self,
+        *,
+        event_bus: EventBus | None = None,
+        database_url: str | None = None,
+        global_invariant_repo: Any | None = None,
+    ) -> None:
+        """Initialise the invariant service.
 
         Args:
             event_bus: Optional EventBus for domain event emission.
+            database_url: SQLAlchemy database URL for persistence (action/plan
+                scope invariants).  When provided, all CRUD operations use the
+                database.  When ``None``, operates in pure in-memory mode.
+            global_invariant_repo: Optional repository for global/project-
+                scoped invariant persistence (Bug #8573).  When provided,
+                add/list/remove operations delegate to this repo instead of
+                building ad-hoc sessions.  Useful when DI container manages
+                the repository lifecycle.
         """
         self._invariants: dict[str, Invariant] = {}
         self._enforcement_records: list[InvariantEnforcementRecord] = []
@@ -60,11 +85,80 @@ class InvariantService:
         self._sanitizer = PromptSanitizer()
         self._event_bus = event_bus
 
+        # Repository-based path (Bug #8573)
+        self._global_invariant_repo = global_invariant_repo
+
+        # Legacy session-factory path (fallback for standalone usage)
+        self._database_url = database_url
+        self._session_factory: sessionmaker[Session] | None = None
+        self._has_loaded_from_db: bool = False
+
+    # ------------------------------------------------------------------
+    # Session-factory helpers (lazy init)
+    # ------------------------------------------------------------------
+
+    def _ensure_session_factory(self) -> sessionmaker[Session]:
+        """Get or create a lazy session factory from ``database_url``."""
+        if self._session_factory is None and self._database_url is not None:
+            engine = _create_engine(
+                self._database_url,
+                echo=False,
+                future=True,
+                isolation_level="SERIALIZABLE",
+                connect_args={"check_same_thread": False}
+                if self._database_url.startswith("sqlite")
+                else {},
+            )
+            # SQLAlchemy 2.x compatible – use the engine's sessionmaker_class
+            try:
+                # SQLAlchemy >= 2.0
+                from sqlalchemy.orm import sessionmaker
+
+                self._session_factory = sessionmaker(
+                    bind=engine,
+                    expire_on_commit=False,
+                    autoflush=False,
+                    autocommit=False,
+                    class_=Session,
+                )
+            except ImportError:
+                # SQLAlchemy < 2.0 – fallback (shouldn't happen in practice)
+                self._session_factory = engine.sessionmaker(
+                    expire_on_commit=False,
+                    autoflush=False,
+                    autocommit=False,
+                )
+        return self._session_factory  # type: ignore[return-value]
+
+    def _ensure_loaded_from_db(self) -> None:
+        """Populate the in-memory cache from the database (one-shot)."""
+        if (
+            self._database_url is not None
+            and self._session_factory is not None
+            and not self._has_loaded_from_db
+        ):
+            session = self._session_factory()
+            try:
+                from cleveragents.infrastructure.database.models import InvariantModel
+
+                rows = (
+                    session.query(InvariantModel)
+                    .filter(InvariantModel.active == True)  # noqa: E712
+                    .all()
+                )
+                for row in rows:
+                    domain_inv = row.to_domain()
+                    self._invariants[domain_inv.id] = domain_inv
+                self._has_loaded_from_db = True
+            finally:
+                session.close()
+
+    # ------------------------------------------------------------------
+    # Public API
+    # ------------------------------------------------------------------
+
     def add_invariant(
-        self,
-        text: str,
-        scope: InvariantScope,
-        source_name: str,
+        self, text: str, scope: InvariantScope, source_name: str
     ) -> Invariant:
         """Add a new invariant with validation.
 
@@ -77,14 +171,13 @@ class InvariantService:
             The created ``Invariant``.
 
         Raises:
-            ValidationError: If text is empty/blank or source_name is blank.
+            ValidationError: If text is empty / blank or source_name is blank.
         """
         if not text or not text.strip():
             raise ValidationError("Invariant text must not be empty")
         if not source_name or not source_name.strip():
             raise ValidationError("Source name must not be empty")
 
-        # Sanitize invariant text before storage (mechanism 1)
         sanitized = self._sanitizer.sanitize_user_input(text.strip())
         text = sanitized.sanitized
 
@@ -95,8 +188,27 @@ class InvariantService:
             source_name=source_name.strip(),
         )
 
+        # Delegate to DI-wired repository for global/project scope (Bug #8573)
+        if self._global_invariant_repo is not None and scope in (
+            InvariantScope.GLOBAL,
+            InvariantScope.PROJECT,
+        ):
+            self._global_invariant_repo.create(invariant)
+        elif self._database_url is not None and self._ensure_session_factory() is not None:
+            session = self._session_factory()
+            try:
+                from cleveragents.infrastructure.database.models import InvariantModel
+
+                model = InvariantModel.from_domain(invariant)
+                session.add(model)
+                session.flush()
+            finally:
+                session.close()
+
+        # Cache in memory regardless of storage path
         self._invariants[invariant.id] = invariant
-        self._logger.info(
+
+        logger.info(
             "Invariant added",
             invariant_id=invariant.id,
             scope=scope.value,
@@ -113,9 +225,9 @@ class InvariantService:
         """Filter and list invariants.
 
         Args:
-            scope: Filter by scope (None = all scopes).
-            source_name: Filter by source name (None = all sources).
-            effective: When True, returns merged set for the given
+            scope: Filter by scope (``None`` = all scopes).
+            source_name: Filter by source name (``None`` = all sources).
+            effective: When ``True``, returns merged set for the given
                 scope chain (requires ``scope`` and ``source_name``).
 
         Returns:
@@ -127,6 +239,50 @@ class InvariantService:
                 project_name=source_name if scope == InvariantScope.PROJECT else None,
             )
 
+        # Delegate to DI-wired repository for global/project scope (Bug #8573)
+        if self._global_invariant_repo is not None and scope in (
+            InvariantScope.GLOBAL,
+            InvariantScope.PROJECT,
+        ):
+            result = self._global_invariant_repo.list_scope(scope=scope)
+            # Build / refresh cache
+            for inv in result:
+                self._invariants.setdefault(inv.id, inv)
+            return result
+
+        # Database-backed query path (one-shot cache population)
+        if (
+            self._database_url is not None
+            and self._ensure_session_factory() is not None
+        ):
+            session = self._session_factory()
+            try:
+                from cleveragents.infrastructure.database.models import InvariantModel
+
+                query = session.query(InvariantModel).filter(
+                    InvariantModel.active == True  # noqa: E712
+                )
+
+                if scope is not None:
+                    query = query.filter(InvariantModel.scope == scope.value)
+
+                if source_name is not None:
+                    query = query.filter(InvariantModel.source_name == source_name)
+
+                rows = query.all()
+                result = [row.to_domain() for row in rows]
+
+                # Build / refresh cache
+                for inv in result:
+                    self._invariants.setdefault(inv.id, inv)
+
+                return result
+            finally:
+                session.close()
+
+        # Pure-in-memory fallback
+        self._ensure_loaded_from_db()  # one-shot pull on demand
+
         result = [inv for inv in self._invariants.values() if inv.active]
 
         if scope is not None:
@@ -138,13 +294,13 @@ class InvariantService:
         return result
 
     def remove_invariant(self, invariant_id: str) -> Invariant:
-        """Soft-delete an invariant by setting active=False.
+        """Soft-delete an invariant by setting ``active=False``.
 
         Args:
             invariant_id: The ULID of the invariant to remove.
 
         Returns:
-            The updated ``Invariant``.
+            The updated (deactivated) ``Invariant``.
 
         Raises:
             NotFoundError: If the invariant does not exist.
@@ -152,22 +308,84 @@ class InvariantService:
         if not invariant_id or not invariant_id.strip():
             raise ValidationError("Invariant ID must not be empty")
 
-        inv = self._invariants.get(invariant_id)
+        # Look up current state from cache or DB
+        inv = self._get_invariant_by_id(invariant_id)
         if inv is None:
             raise NotFoundError(
                 resource_type="invariant",
                 resource_id=invariant_id,
             )
-        # Invariant is frozen (immutable); create a new instance with active=False
-        deactivated = inv.model_copy(update={"active": False})
-        self._invariants[invariant_id] = deactivated
-        self._logger.info("Invariant removed (soft-delete)", invariant_id=invariant_id)
-        return deactivated
+
+        inactive = inv.model_copy(update={"active": False})
+
+        # Persist to DB when configured
+        # Delegate to DI-wired repository for global/project scope (Bug #8573)
+        if self._global_invariant_repo is not None:
+            removed = self._global_invariant_repo.remove(invariant_id)
+            if not removed:
+                raise NotFoundError(
+                    resource_type="invariant",
+                    resource_id=invariant_id,
+                )
+        elif self._database_url is not None and self._session_factory is not None:
+            session = self._session_factory()
+            try:
+                from cleveragents.infrastructure.database.models import InvariantModel
+
+                # Use filter_by instead of deprecated .get()
+                row = (
+                    session.query(InvariantModel)
+                    .filter(InvariantModel.id == invariant_id)
+                    .first()
+                )
+                if row is None:
+                    raise NotFoundError(
+                        resource_type="invariant",
+                        resource_id=invariant_id,
+                    )
+                row.active = False
+                session.flush()
+            finally:
+                session.close()
+
+        self._invariants[invariant_id] = inactive
+        logger.info("Invariant removed (soft-delete)", invariant_id=invariant_id)
+        return inactive
+
+    def _get_invariant_by_id(self, invariant_id: str) -> Invariant | None:
+        """Lookup a single invariant by ID (cache, repo, or DB)."""
+        if invariant_id in self._invariants:
+            return self._invariants[invariant_id]
+
+        # Check DI-wired repository first (Bug #8573)
+        if self._global_invariant_repo is not None:
+            inv = self._global_invariant_repo.get_by_id(invariant_id)
+            if inv is not None:
+                self._invariants[invariant_id] = inv
+                return inv
+
+        # Fallback to legacy session-factory path
+        if self._database_url is not None and self._session_factory is not None:
+            session = self._session_factory()
+            try:
+                from cleveragents.infrastructure.database.models import InvariantModel
+
+                row = (
+                    session.query(InvariantModel)
+                    .filter_by(id=invariant_id)
+                    .first()
+                )
+                if row is not None:
+                    inv = row.to_domain()
+                    self._invariants[invariant_id] = inv
+                    return inv
+            finally:
+                session.close()
+
+        return None
 
     def get_effective_invariants(
-        self,
-        plan_id: str | None = None,
-        project_name: str | None = None,
+        self, plan_id: str | None = None, project_name: str | None = None
     ) -> list[Invariant]:
         """Return the merged precedence chain for a plan/project context.
 
@@ -175,10 +393,8 @@ class InvariantService:
         using plan > project > global precedence.
 
         Args:
-            plan_id: Optional plan identifier to collect plan-scoped
-                invariants.
-            project_name: Optional project name to collect project-scoped
-                invariants.
+            plan_id: Optional plan identifier to collect plan-scoped invariants.
+            project_name: Optional project name to collect project-scoped invariants.
 
         Returns:
             Merged, de-duplicated list of effective invariants.
@@ -257,7 +473,7 @@ class InvariantService:
                         )
                     )
                 except Exception:
-                    self._logger.warning(
+                    self._logger.warning(  # type: ignore[attr-defined]
                         "event_bus_emit_failed",
                         event_type="INVARIANT_VIOLATED",
                         plan_id=plan_id,
@@ -285,7 +501,7 @@ class InvariantService:
                         )
                     )
                 except Exception:
-                    self._logger.warning(
+                    self._logger.warning(  # type: ignore[attr-defined]
                         "event_bus_emit_failed",
                         event_type="INVARIANT_ENFORCED",
                         plan_id=plan_id,
@@ -303,7 +519,7 @@ class InvariantService:
                     )
                 )
             except Exception:
-                self._logger.warning(
+                self._logger.warning(  # type: ignore[attr-defined]
                     "event_bus_emit_failed",
                     event_type="INVARIANT_RECONCILED",
                     plan_id=plan_id,

src/cleveragents/application/services/subplan_service.py

@@ -7,6 +7,17 @@ presence, and ``max_parallel`` bounds before creating child plans.
 Spawn metadata (spawn decision ID, parent/root plan IDs, execution mode) is
 persisted alongside each child plan for status output and provenance tracking.
 
+Invariant propagation
+---------------------
+
+Per the specification (Glossary → Invariant): "recorded as
+``invariant_enforced`` decisions that propagate to child plans."
+
+When a child plan is spawned, all ``invariant_enforced`` decisions from the
+parent plan's decision tree are re-recorded on the child plan's decision tree.
+This ensures that child plans start Strategize with the full invariant set
+enforced on the parent, including ``non_overridable`` global invariants.
+
 Design decisions:
     - Dependency injection: ``DecisionService`` and ``UnitOfWork`` are
       injected via the constructor (consistent with existing service patterns).
@@ -19,6 +30,7 @@ Based on:
     - docs/specification.md L18170-L18295 (subplan spawning)
     - ADR-006 (Plan Lifecycle)
     - Forgejo issue #197
+    - Forgejo issue #9131 (invariant_enforced propagation fix)
 """
 
 from __future__ import annotations
@@ -154,6 +166,8 @@ class SubplanService:
     2. Validating each entry (resource scopes, merge strategy, parallelism).
     3. Building ``SubplanStatus`` objects for the parent plan.
     4. Persisting ``SpawnMetadata`` for status queries.
+    5. Propagating ``invariant_enforced`` decisions from the parent plan to
+       each child plan's decision tree (spec: Glossary → Invariant).
 
     Args:
         decision_service: Service for querying decision records.
@@ -193,6 +207,12 @@ class SubplanService:
         and ``SpawnMetadata`` for each entry.  The caller is responsible for
         persisting the updated parent plan.
 
+        Invariant propagation: all ``invariant_enforced`` decisions from the
+        parent plan are re-recorded on each child plan's decision tree so that
+        child plans start Strategize with the parent's full invariant set.
+        This satisfies the spec requirement: "recorded as ``invariant_enforced``
+        decisions that propagate to child plans."
+
         Args:
             parent_plan: The parent plan that will own the child plans.
             config: Subplan execution configuration.
@@ -228,8 +248,17 @@ class SubplanService:
         if not validation.valid:
             raise SpawnValidationError(validation.errors)
 
-        # Build statuses, metadata, and real child Plan objects
+        # Retrieve parent plan's invariant_enforced decisions once, before
+        # creating child plans.  These will be propagated to every child.
         parent_id: str = parent_plan.identity.plan_id
+        parent_invariant_decisions: list[Decision] = (
+            self._decision_service.list_by_type(
+                parent_id,
+                DecisionType.INVARIANT_ENFORCED,
+            )
+        )
+
+        # Build statuses, metadata, and real child Plan objects
         root_id: str = parent_plan.identity.root_plan_id or parent_id
         mode: str = config.execution_mode.value
 
@@ -283,6 +312,16 @@ class SubplanService:
             )
             child_plans.append(child_plan)
 
+            # Propagate invariant_enforced decisions from parent to child.
+            # Each parent invariant decision is re-recorded on the child plan
+            # so that the child starts Strategize with the same invariant set.
+            # Spec: Glossary → Invariant: "recorded as invariant_enforced
+            # decisions that propagate to child plans."
+            self._propagate_invariant_decisions(
+                child_plan_id=subplan_id,
+                parent_invariant_decisions=parent_invariant_decisions,
+            )
+
         # Attach subplan statuses to the parent plan for lifecycle tracking
         parent_plan.subplan_statuses = list(parent_plan.subplan_statuses) + statuses
 
@@ -293,6 +332,7 @@ class SubplanService:
                 "root_plan_id": root_id,
                 "count": len(statuses),
                 "execution_mode": mode,
+                "invariant_decisions_propagated": len(parent_invariant_decisions),
             },
         )
 
@@ -304,6 +344,40 @@ class SubplanService:
             child_plans=child_plans,
         )
 
+    # ------------------------------------------------------------------
+    # _propagate_invariant_decisions
+    # ------------------------------------------------------------------
+
+    def _propagate_invariant_decisions(
+        self,
+        child_plan_id: str,
+        parent_invariant_decisions: list[Decision],
+    ) -> None:
+        """Propagate invariant_enforced decisions from parent to child plan.
+
+        Re-records each parent ``invariant_enforced`` decision on the child
+        plan's decision tree.  The child plan starts Strategize with the same
+        invariant set as the parent, satisfying the spec requirement:
+        "recorded as ``invariant_enforced`` decisions that propagate to child
+        plans."
+
+        Args:
+            child_plan_id: ULID of the child plan to record decisions on.
+            parent_invariant_decisions: The parent plan's ``invariant_enforced``
+                decisions to propagate.
+        """
+        for parent_decision in parent_invariant_decisions:
+            self._decision_service.record_decision(
+                plan_id=child_plan_id,
+                decision_type=DecisionType.INVARIANT_ENFORCED,
+                question=parent_decision.question,
+                chosen_option=parent_decision.chosen_option,
+                rationale=parent_decision.rationale,
+                context_snapshot=parent_decision.context_snapshot,
+                confidence_score=parent_decision.confidence_score,
+                alternatives_considered=list(parent_decision.alternatives_considered),
+            )
+
     # ------------------------------------------------------------------
     # validate_spawn
     # ------------------------------------------------------------------

src/cleveragents/cli/commands/invariant.py

@@ -43,6 +43,7 @@ import typer
 from rich.console import Console
 from rich.table import Table
 
+from cleveragents.application.container import get_container
 from cleveragents.application.services.invariant_service import InvariantService
 from cleveragents.cli.formatting import OutputFormat, format_output
 from cleveragents.core.exceptions import CleverAgentsError, NotFoundError
@@ -55,16 +56,19 @@ console = Console()
 
 _FORMAT_HELP = "Output format: json, yaml, plain, table, or rich (default: rich)"
 
-# Module-level service instance (in-memory, same lifetime as CLI process)
-_service: InvariantService | None = None
 
+def _get_service() -> InvariantService | None:
+    """Return the DI-wired InvariantService for the current process.
 
-def _get_service() -> InvariantService:
-    """Return (or lazily create) the module-level InvariantService."""
-    global _service
-    if _service is None:
-        _service = InvariantService()
-    return _service
+    Uses the global container so that the invariant service is properly wired
+    with its repository dependency (Bug #8573).  Returns ``None`` when no
+    container is available (e.g. in tests without a DB).
+    """
+    try:
+        container = get_container()
+        return container.invariant_service()
+    except Exception:
+        return None
 
 
 def _resolve_scope(
@@ -131,6 +135,11 @@ def add(
     try:
         scope, source_name = _resolve_scope(is_global, project, plan, action)
         service = _get_service()
+        if service is None:
+            console.print(
+                "[red]Error:[/red] Invariant service not available (no database?)"
+            )
+            raise typer.Abort()
         inv = service.add_invariant(text=text, scope=scope, source_name=source_name)
 
         if fmt != OutputFormat.RICH.value:
@@ -180,6 +189,11 @@ def list_invariants(
     """
     try:
         service = _get_service()
+        if service is None:
+            console.print(
+                "[red]Error:[/red] Invariant service not available (no database?)"
+            )
+            raise typer.Abort()
 
         scope: InvariantScope | None = None
         source_name: str | None = None
@@ -265,6 +279,11 @@ def remove(
                 raise typer.Abort()
 
         service = _get_service()
+        if service is None:
+            console.print(
+                "[red]Error:[/red] Invariant service not available (no database?)"
+            )
+            raise typer.Abort()
         inv = service.remove_invariant(invariant_id)
 
         if fmt != OutputFormat.RICH.value:

src/cleveragents/cli/commands/plan.py

@@ -627,6 +627,19 @@ def _create_sandbox_for_plan(
 
     container = get_container()
     plan = service.get_plan(plan_id)
+
+    # Guard: when plan is already execute/processing or execute/complete,
+    # the sandbox branch holds output awaiting apply or is actively being
+    # used by an in-progress execution.  Do NOT destroy it via cleanup_stale.
+    if (
+        plan is not None
+        and plan.phase == PlanPhase.EXECUTE
+        and plan.state in (ProcessingState.PROCESSING, ProcessingState.COMPLETE)
+    ):
+        flat_root = os.path.join(os.getcwd(), ".cleveragents", "sandbox")
+        os.makedirs(flat_root, exist_ok=True)
+        return flat_root, []
+
     project_names = [pl.project_name for pl in getattr(plan, "project_links", [])]
 
     sandboxes: list[_SandboxInfo] = []

src/cleveragents/cli/main.py

@@ -3,6 +3,7 @@
 Based on ADR-009: CLI Framework using Typer.
 """
 
+import os
 import sys
 from pathlib import Path
 from typing import TYPE_CHECKING, Annotated, Any
@@ -249,6 +250,13 @@ def _print_basic_help() -> None:
     """Print a lightweight help message without heavy imports."""
     typer.echo("CleverAgents - AI-powered development assistant (actor-first)")
     typer.echo("Usage: cleveragents [OPTIONS] COMMAND [ARGS]...")
+    typer.echo("\nGlobal options:")
+    typer.echo("  --data-dir PATH    Override global data directory")
+    typer.echo("  --config-path PATH Override global configuration file path")
+    typer.echo("  -v                 Increase log verbosity (repeatable)")
+    typer.echo("  --format -f FMT    Output format: rich/color/table/plain/json/yaml")
+    typer.echo("  --version          Show version and exit")
+    typer.echo("  --help -h          Show this message and exit")
     typer.echo("\nCommon commands:")
     typer.echo("  project      Project management")
     typer.echo("  actor context  Actor context management")
@@ -272,6 +280,21 @@ def _print_basic_help() -> None:
     )
 
 
+# ---------------------------------------------------------------------------
+# Verbosity (-v) → log level mapping (ADR-021 §Global CLI Flags)
+# ---------------------------------------------------------------------------
+# Mapping: 0 = silent (CRITICAL), 1 = ERROR, 2 = WARNING, 3 = INFO,
+#          4 = DEBUG, 5+ = DEBUG (TRACE not available in Python stdlib).
+_VERBOSITY_LOG_LEVELS: tuple[str, ...] = (
+    "CRITICAL",  # 0 — no -v flag (silent)
+    "ERROR",  # 1 — -v
+    "WARNING",  # 2 — -vv
+    "INFO",  # 3 — -vvv
+    "DEBUG",  # 4 — -vvvv
+    "DEBUG",  # 5+ — -vvvvv (TRACE mapped to DEBUG)
+)
+
+
 def version_callback(value: bool) -> None:
     """Handle --version flag."""
     if value:
@@ -316,20 +339,98 @@ def main_callback(
             ),
         ),
     ] = OutputFormat.RICH,
+    data_dir: Annotated[
+        Path | None,
+        typer.Option(
+            "--data-dir",
+            help=(
+                "Override the global data directory for this invocation "
+                "(database, caches, sessions, logs). "
+                "Overrides CLEVERAGENTS_DATA_DIR and core.data-dir config key."
+            ),
+            metavar="PATH",
+        ),
+    ] = None,
+    config_path: Annotated[
+        Path | None,
+        typer.Option(
+            "--config-path",
+            help=(
+                "Override the global configuration file path for this invocation. "
+                "Overrides CLEVERAGENTS_CONFIG_PATH and the default config location."
+            ),
+            metavar="PATH",
+        ),
+    ] = None,
+    verbose: Annotated[
+        int,
+        typer.Option(
+            "-v",
+            count=True,
+            help=(
+                "Increase log verbosity (repeatable). "
+                "No flag = silent; -v = ERROR; -vv = WARN; "
+                "-vvv = INFO; -vvvv = DEBUG; -vvvvv = TRACE."
+            ),
+        ),
+    ] = 0,
 ) -> None:
     """CleverAgents - AI-powered development assistant."""
-    # Suppress debug-level logs on stdout for ALL commands so machine-readable
-    # output formats (json, yaml, plain) receive clean stdout.  Commands that
-    # need verbose logging can override this after parsing --log-level flags.
     from cleveragents.config.logging import configure_structlog
 
-    configure_structlog(log_level="WARNING")
+    # -----------------------------------------------------------------------
+    # Validate and wire --data-dir
+    # -----------------------------------------------------------------------
+    if data_dir is not None:
+        resolved_data_dir = data_dir.resolve()
+        if resolved_data_dir.exists() and not resolved_data_dir.is_dir():
+            get_err_console().print(
+                f"[red]Error: --data-dir '{data_dir}' exists but is not"
+                " a directory.[/red]"
+            )
+            raise typer.Exit(1)
+        # Wire: set env var so Settings and other components pick it up.
+        # Settings.__new__ reads CLEVERAGENTS_DATA_DIR on first access, so
+        # setting the env var before any Settings-reading code runs is the
+        # correct (and non-destructive) way to propagate the CLI override.
+        os.environ["CLEVERAGENTS_DATA_DIR"] = str(resolved_data_dir)
+
+    # -----------------------------------------------------------------------
+    # Validate and wire --config-path
+    # -----------------------------------------------------------------------
+    if config_path is not None:
+        resolved_config_path = config_path.resolve()
+        if not resolved_config_path.exists():
+            get_err_console().print(
+                f"[red]Error: --config-path '{config_path}' does not exist.[/red]"
+            )
+            raise typer.Exit(1)
+        if not resolved_config_path.is_file():
+            get_err_console().print(
+                f"[red]Error: --config-path '{config_path}' is not a file.[/red]"
+            )
+            raise typer.Exit(1)
+        # Wire: set env var so ConfigService instances created later use this path
+        os.environ["CLEVERAGENTS_CONFIG_PATH"] = str(resolved_config_path)
+
+    # -----------------------------------------------------------------------
+    # Configure log verbosity from -v count (ADR-021 §Global CLI Flags)
+    # -----------------------------------------------------------------------
+    log_level = _VERBOSITY_LOG_LEVELS[min(verbose, len(_VERBOSITY_LOG_LEVELS) - 1)]
+    configure_structlog(log_level=log_level)
+
     _register_subcommands()
 
-    # Store the selected output format in the Typer context so all subcommands
-    # can read it via ctx.obj["format"] without needing their own --format flag.
+    # -----------------------------------------------------------------------
+    # Store all global options in ctx.obj for subcommand access
+    # -----------------------------------------------------------------------
     ctx.ensure_object(dict)
     ctx.obj["format"] = fmt.value
+    ctx.obj["data_dir"] = str(data_dir.resolve()) if data_dir is not None else None
+    ctx.obj["config_path"] = (
+        str(config_path.resolve()) if config_path is not None else None
+    )
+    ctx.obj["verbose"] = verbose
 
 
 @app.command()
@@ -760,8 +861,13 @@ def main(args: list[str] | None = None) -> int:
         return 130
     except Exception as e:
         from cleveragents.core.error_handling import classify_error, wrap_unexpected
+        from cleveragents.shared.redaction import redact_value
 
         err_console = get_err_console()
+        # Always print the original exception type/message to stderr so that
+        # actionable details (e.g. "No such option: --flag") remain visible
+        # even when the log level is set to CRITICAL (structlog suppressed).
+        err_console.print(f"[dim]{type(e).__name__}: {redact_value(str(e))}[/dim]")
         safe = wrap_unexpected(e)
         info = classify_error(safe)
         err_console.print(

src/cleveragents/domain/repositories/invariant_repository.py

@@ -0,0 +1,76 @@
+"""Domain repository protocol for standalone invariant constraints.
+
+Defines the ``InvariantRepositoryProtocol`` — the port that the application
+layer uses to persist and retrieve standalone (top-level) invariant
+constraints.  Infrastructure adapters (e.g. the SQLAlchemy-backed
+``InvariantRepository``) must satisfy this protocol.
+
+Based on the clean architecture principle described in the specification:
+adapters live at the edge; the domain layer defines the contracts.
+
+Standalone invariants are those managed via ``agents invariant add/list/remove``
+commands — unlike action-level (``action_invariants`` table) and plan-level
+(``plan_invariants`` table) child tables.
+"""
+
+from __future__ import annotations
+
+from typing import Protocol, runtime_checkable
+
+from cleveragents.domain.models.core.invariant import Invariant
+
+
+@runtime_checkable
+class InvariantRepositoryProtocol(Protocol):
+    """Port for standalone invariant persistence.
+
+    All methods that mutate state flush but do **not** commit; the caller
+    or a Unit-of-Work wrapper is responsible for committing the transaction.
+    """
+
+    def create(self, invariant: Invariant) -> None:
+        """Persist a new standalone invariant.
+
+        Args:
+            invariant: The ``Invariant`` domain model to persist.
+        """
+        ...
+
+    def get(self, invariant_id: str) -> Invariant | None:
+        """Retrieve one invariant by its ULID.
+
+        Args:
+            invariant_id: ULID string of the invariant.
+
+        Returns:
+            The ``Invariant`` domain model, or ``None`` if not found.
+        """
+        ...
+
+    def list_invariants(
+        self,
+        scope: str | None = None,
+        source_name: str | None = None,
+        active_only: bool = True,
+    ) -> list[Invariant]:
+        """List invariants with optional filters.
+
+        Args:
+            scope: Filter by scope value ('global', 'project', 'action', 'plan').
+            source_name: Filter by source name.
+            active_only: If ``True``, only return active (non-deleted) invariants.
+
+        Returns:
+            List of ``Invariant`` domain models.
+        """
+        ...
+
+    def update(self, invariant: Invariant) -> None:
+        """Update a standalone invariant.
+
+        Used primarily for soft-delete (setting ``active`` to ``False``).
+
+        Args:
+            invariant: The updated ``Invariant`` (same id as persisted record).
+        """
+        ...

src/cleveragents/infrastructure/database/invariant_repository.py

@@ -0,0 +1,145 @@
+"""SQLAlchemy-backed repository for standalone invariant constraints.
+
+Implements :class:`~cleveragents.domain.repositories.invariant_repository.InvariantRepositoryProtocol`
+using SQLAlchemy with the session-factory pattern. Operations flush but do NOT
+commit — callers are responsible for committing transactions.
+
+Based on ADR-007 (Repository Pattern) and Phase 0 discovery.
+Includes retry logic per ADR-033.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from typing import TYPE_CHECKING
+
+import structlog
+from sqlalchemy.orm import Session
+
+from cleveragents.core.exceptions import NotFoundError
+from cleveragents.core.retry_patterns import (
+    retry_database_operation as database_retry,
+)
+
+if TYPE_CHECKING:
+    from cleveragents.domain.models.core.invariant import InvariantScope
+
+logger = structlog.get_logger(__name__)
+
+
+class InvariantRepository:
+    """Repository for standalone (top-level) invariant persistence.
+
+    Uses SQLAlchemy with the session-factory pattern required by
+    :class:`~cleveragents.infrastructure.database.repositories` peers.
+    Operations flush but do not commit.
+    """
+
+    def __init__(self, session_factory: Callable[[], Session]) -> None:
+        """Initialize repository with a session factory.
+
+        Args:
+            session_factory: Callable returning a new SQLAlchemy ``Session``.
+        """
+        self._session_factory = session_factory
+
+    @database_retry
+    def create(self, invariant: "Invariant") -> None:
+        """Persist a new standalone invariant to the database."""
+        from cleveragents.infrastructure.database.models import InvariantModel
+
+        session = self._session_factory()
+        try:
+            model = InvariantModel.from_domain(invariant)
+            session.add(model)
+            session.flush()
+            logger.info(
+                "Invariant persisted",
+                invariant_id=invariant.id,
+                scope=invariant.scope.value,
+                source_name=invariant.source_name,
+            )
+        except Exception:
+            session.rollback()
+            raise
+        finally:
+            session.close()
+
+    @database_retry
+    def get(self, invariant_id: str) -> "Invariant | None":
+        """Retrieve one invariant by its ULID."""
+        from cleveragents.infrastructure.database.models import InvariantModel
+
+        session = self._session_factory()
+        try:
+            row = session.query(InvariantModel).get(invariant_id)
+            if row is None:
+                return None
+            return row.to_domain()
+        finally:
+            session.close()
+
+    @database_retry
+    def list_invariants(
+        self,
+        scope: "InvariantScope | str | None" = None,
+        source_name: str | None = None,
+        active_only: bool = True,
+    ) -> list["Invariant"]:
+        """List invariants with optional filters."""
+        from cleveragents.infrastructure.database.models import InvariantModel
+        from cleveragents.domain.models.core.invariant import InvariantScope
+
+        session = self._session_factory()
+        try:
+            query = session.query(InvariantModel)
+
+            if scope is not None:
+                scope_val = scope.value if isinstance(scope, InvariantScope) else scope
+                query = query.filter(InvariantModel.scope == scope_val)
+
+            if source_name is not None:
+                query = query.filter(InvariantModel.source_name == source_name)
+
+            if active_only:
+                query = query.filter(InvariantModel.active == True)  # noqa: E712
+
+            rows = query.all()
+            return [row.to_domain() for row in rows]
+        finally:
+            session.close()
+
+    @database_retry
+    def update(self, invariant: "Invariant") -> None:
+        """Update a standalone invariant (used for soft-delete)."""
+        from cleveragents.infrastructure.database.models import InvariantModel
+
+        session = self._session_factory()
+        try:
+            model = session.query(InvariantModel).get(invariant.id)
+            if model is None:
+                raise NotFoundError(
+                    resource_type="invariant",
+                    resource_id=invariant.id,
+                )
+
+            # Update mutable fields
+            model.text = invariant.text
+            model.scope = invariant.scope.value
+            model.source_name = invariant.source_name
+            model.active = invariant.active
+            model.non_overridable = invariant.non_overridable
+            model.created_at = invariant.created_at.isoformat()
+
+            session.flush()
+            logger.info(
+                "Invariant updated",
+                invariant_id=invariant.id,
+            )
+        except NotFoundError:
+            raise
+        except Exception:
+            session.rollback()
+            raise
+        finally:
+            session.close()

src/cleveragents/infrastructure/database/migrations/versions/m11_001_standalone_invariants.py

@@ -0,0 +1,62 @@
+"""Create standalone invariants table for InvariantService persistence.
+
+Adds a new ``invariants`` table for top-level invariant constraints managed
+by :class:`~cleveragents.application.services.invariant_service.InvariantService`
+via the CLI ``agents invariant add/list/remove`` commands.  This table is
+separate from action-level (``action_invariants``) and plan-level
+(``plan_invariants``) child tables, allowing invariants to persist across
+CLI invocations even when not attached to a specific action or plan.
+
+Revision ID: m11_001_standalone_invariants
+Revises: m9_003_plan_result_success_column
+Create Date: 2026-05-12 00:00:00
+
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers, used by Alembic.
+revision: str = "m11_001_standalone_invariants"
+down_revision: str | None = "m9_003_plan_result_success_column"
+branch_labels: str | None = None
+depends_on: str | None = None
+
+
+def upgrade() -> None:
+    """Create the standalone invariants table and indexes."""
+    op.create_table(
+        "invariants",
+        sa.Column("id", sa.String(26), primary_key=True, nullable=False),
+        sa.Column("text", sa.Text(), nullable=False),
+        sa.Column("scope", sa.String(20), nullable=False),
+        sa.Column("source_name", sa.String(255), nullable=False),
+        sa.Column(
+            "created_at",
+            sa.String(30),
+            nullable=False,
+            server_default=sa.func.datetime("now"),
+        ),
+        sa.Column(
+            "active", sa.Boolean(), nullable=False, server_default=sa.text("1")
+        ),
+        sa.Column(
+            "non_overridable",
+            sa.Boolean(),
+            nullable=False,
+            server_default=sa.text("0"),
+        ),
+        sa.CheckConstraint(
+            "scope IN ('global', 'project', 'action', 'plan')",
+            name="ck_invariants_scope",
+        ),
+    )
+    op.create_index("ix_invariants_scope", "invariants", ["scope"])
+    op.create_index("ix_invariants_source_name", "invariants", ["source_name"])
+
+
+def downgrade() -> None:
+    """Drop the standalone invariants table and indexes."""
+    op.drop_index("ix_invariants_source_name", table_name="invariants")
+    op.drop_index("ix_invariants_scope", table_name="invariants")
+    op.drop_table("invariants")

src/cleveragents/infrastructure/database/models.py

@@ -14,6 +14,7 @@ Alembic migrations.
 | ``plan_projects``      | ``PlanProjectModel``       | Plan-project links     |
 | ``plan_arguments``     | ``PlanArgumentModel``      | Plan argument values   |
 | ``plan_invariants``    | ``PlanInvariantModel``     | Plan invariant rules   |
+| ``invariants``         | ``InvariantModel``         | Standalone invariants    |
 | ``resource_types``     | ``ResourceTypeModel``      | Resource type defs     |
 | ``resources``          | ``ResourceModel``          | Resource instances     |
 | ``resource_edges``     | ``ResourceEdgeModel``      | Resource DAG edges     |
@@ -37,6 +38,7 @@ import json
 import logging
 import re
 from datetime import UTC, datetime
+from ulid import ULID
 from typing import TYPE_CHECKING, Any, cast
 
 if TYPE_CHECKING:
@@ -59,7 +61,7 @@ from sqlalchemy import (
     Text,
     UniqueConstraint,
     create_engine,
-    text,
+    text as sa_text,
 )
 from sqlalchemy.orm import (
     Mapped,
@@ -1772,7 +1774,7 @@ class ResourceLinkModel(Base):  # type: ignore[misc]
         Text,
         nullable=False,
         default="contains",
-        server_default=text("'contains'"),
+        server_default=sa_text("'contains'"),
     )
 
     # Timestamp (ISO-8601 string)
@@ -2841,8 +2843,8 @@ class DecisionModel(Base):  # type: ignore[misc]
         Index(
             "idx_decisions_superseded",
             "superseded_by",
-            postgresql_where=text("superseded_by IS NOT NULL"),
-            sqlite_where=text("superseded_by IS NOT NULL"),
+            postgresql_where=sa_text("superseded_by IS NOT NULL"),
+            sqlite_where=sa_text("superseded_by IS NOT NULL"),
         ),
     )
 
@@ -3268,7 +3270,7 @@ class CorrectionAttemptModel(Base):  # type: ignore[misc]
     created_at = Column(
         String(30),
         nullable=False,
-        server_default=text("(strftime('%Y-%m-%dT%H:%M:%f', 'now'))"),
+        server_default=sa_text("(strftime('%Y-%m-%dT%H:%M:%f', 'now'))"),
     )
     completed_at = Column(String(30), nullable=True)
 
@@ -3392,6 +3394,110 @@ class CorrectionAttemptModel(Base):  # type: ignore[misc]
         )
 
 
+
+
+# ---------------------------------------------------------------------------
+# Standalone Invariant Model (Stage M3.5 - migration m11_001)
+# ---------------------------------------------------------------------------
+
+
+class InvariantModel(Base):  # type: ignore[misc]
+    """Database model for standalone invariant constraints.
+
+    Stores top-level invariant rules managed by :class:`~cleveragents.application.services.invariant_service.InvariantService`.
+    Separate from action-level (``action_invariants``) and plan-level
+    (``plan_invariants``) child tables, allowing invariants to persist
+    across CLI invocations even when not attached to a specific action or plan.
+
+    Table: ``invariants``
+    """
+
+    __allow_unmapped__ = True
+    __tablename__ = "invariants"
+
+    # PK: ULID (26-char string)
+    id = Column(String(26), primary_key=True)
+
+    # Constraint text
+    text = Column(Text, nullable=False)
+
+    # Scope: global | project | action | plan
+    scope = Column(String(20), nullable=False)
+
+    # Owning source (project name, action namespaced name, or "system" for global)
+    source_name = Column(String(255), nullable=False)
+
+    # Soft-delete flag (active vs archived)
+    active = Column(Boolean, nullable=False, default=True, server_default=sa_text("1"))
+
+    # Whether this invariant cannot be overridden by project-level invariants
+    non_overridable = Column(
+        Boolean, nullable=False, default=False, server_default=sa_text("0"),
+    )
+
+    # Timestamp (ISO-8601 string)
+    created_at = Column(String(30), nullable=False)
+
+    __table_args__ = (
+        CheckConstraint(
+            "scope IN ('global', 'project', 'action', 'plan')",
+            name="ck_invariants_scope",
+        ),
+        Index("ix_invariants_scope", "scope"),
+        Index("ix_invariants_source_name", "source_name"),
+    )
+
+    # -- Domain conversion helpers ------------------------------------------
+
+    def to_domain(self) -> Any:
+        """Convert to ``Invariant`` domain model.
+
+        Returns:
+            An ``Invariant`` domain instance.
+        """
+        from cleveragents.domain.models.core.invariant import Invariant, InvariantScope
+
+        return Invariant(
+            id=cast(str, self.id),
+            text=cast(str, self.text),
+            scope=InvariantScope(cast(str, self.scope)),
+            source_name=cast(str, self.source_name),
+            created_at=datetime.fromisoformat(cast(str, self.created_at)),
+            active=self.active,
+        )
+
+    @classmethod
+    def from_domain(cls, invariant: Any) -> InvariantModel:
+        """Create from ``Invariant`` domain model.
+
+        Args:
+            invariant: An ``Invariant`` domain instance.
+
+        Returns:
+            An ``InvariantModel`` ready for persistence.
+        """
+        return cls(
+            id=cast(str, invariant.id),
+            text=invariant.text if hasattr(invariant, "text") else str(invariant),
+            scope=(
+                invariant.scope.value
+                if hasattr(invariant.scope, "value")
+                else str(invariant.scope)
+            ),
+            source_name=(
+                invariant.source_name
+                if hasattr(invariant, "source_name")
+                else ""
+            ),
+            active=getattr(invariant, "active", True),
+            non_overridable=getattr(invariant, "non_overridable", False),
+            created_at=(
+                invariant.created_at.isoformat()
+                if hasattr(invariant, "created_at") and invariant.created_at
+                else datetime.now(tz=UTC).isoformat()
+            ),
+        )
+
 # Database initialization functions
 def init_database(database_url: str = "sqlite:///.cleveragents/db.sqlite") -> Any:
     """Initialize the database.

src/cleveragents/infrastructure/database/repositories.py

@@ -13,6 +13,7 @@ Provides database-backed repositories following the session-factory pattern
 | ``ProjectResourceLinkRepository``| Project-resource link management        |
 | ``DecisionRepository``         | CRUD for decision tree nodes             |
 | ``CorrectionAttemptRepository``| CRUD for correction attempt records      |
+| ``GlobalInvariantRepository``  | Persistence for CLI global/project invariants (Bug #8573) |
 
 ## Session-Factory Pattern
 
@@ -112,6 +113,7 @@ from cleveragents.infrastructure.database.models import (
     CorrectionAttemptModel,
     DebugAttemptModel,
     DecisionModel,
+    InvariantModel,
     LifecycleActionModel,
     LifecyclePlanModel,
     NamespacedProjectModel,
@@ -147,6 +149,7 @@ from cleveragents.domain.models.core.correction import (
     CorrectionAttemptState,
     validate_correction_state_transition,
 )
+from cleveragents.domain.models.core.invariant import InvariantScope
 
 _log = structlog.get_logger(__name__)
 
@@ -6227,8 +6230,158 @@ class CorrectionAttemptRepository:
             session.delete(row)
             session.flush()
             return True
+       except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            session.rollback()
+            raise DatabaseError(
+                f"Failed to delete correction attempt {correction_attempt_id}: {exc}",
+            ) from exc
+
+
+# ---------------------------------------------------------------------------
+# Global Invariant Repository (Bug #8573 - persistence for CLI global/project invariants)
+# ---------------------------------------------------------------------------
+
+
+class GlobalInvariantRepository:
+    """Repository for persisted global and project-scoped invariants.
+
+    Uses the session-factory pattern consistent with other repositories.
+    Unlike action/plan invariants that are children of their parent resources,
+    global and project invariants are standalone rows in the ``invariants``
+    table filtered by scope ('global' | 'project').
+
+    All mutating methods flush but do NOT commit by default; the caller
+    or a ``UnitOfWork`` wrapper is responsible for committing the
+    transaction.  When ``auto_commit`` is ``True`` (e.g. CLI usage
+    outside a UoW), each method commits and closes its own session.
+    """
+
+    def __init__(
+        self,
+        session_factory: Callable[[], Session],
+        *,
+        auto_commit: bool = False,
+    ) -> None:
+        """Initialise with a callable that returns a new SQLAlchemy Session.
+
+        Args:
+            session_factory: Factory returning a new SQLAlchemy ``Session``.
+            auto_commit: When ``True``, each public method commits and
+                closes its session automatically.  Useful for CLI commands
+                that operate outside a ``UnitOfWork``.
+        """
+        self._session_factory = session_factory
+        self._auto_commit = auto_commit
+
+    def _session(self) -> Session:
+        """Convenience helper to obtain a session."""
+        return self._session_factory()
+
+    @database_retry
+    def create(self, invariant: Any) -> Any:
+        """Persist a new global/project ``Invariant`` domain object.
+
+        Args:
+            invariant: A domain ``Invariant`` instance with scope 'global' or
+                'project'.
+
+        Returns:
+            The invariant after persistence.
+
+        Raises:
+            DatabaseError: On transient or unexpected DB errors.
+        """
+        session = self._session()
+        try:
+            db_model = InvariantModel.from_domain(invariant)
+            session.add(db_model)
+            session.flush()
+            if self._auto_commit:
+                session.commit()
+                session.close()
+            return invariant
+        except IntegrityError as exc:
+            session.rollback()
+            raise DatabaseError(f"Failed to create global invariant: {exc}") from exc
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            session.rollback()
+            raise DatabaseError(f"Failed to create global invariant: {exc}") from exc
+
+    @database_retry
+    def list_scope(
+        self,
+        scope: InvariantScope | None = None,
+    ) -> list[Any]:
+        """List global (or project-scoped) invariants.
+
+        Args:
+            scope: Optional scope filter ('global' or 'project').
+
+        Returns:
+            List of ``Invariant`` domain objects.
+        """
+        session = self._session()
+        try:
+            query = session.query(InvariantModel).filter(
+                InvariantModel.active == True,  # noqa: E712
+            )
+            if scope is not None:
+                query = query.filter(InvariantModel.scope == scope.value)
+            rows = query.order_by(InvariantModel.created_at).all()
+            return [row.to_domain() for row in rows]
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            raise DatabaseError(f"Failed to list global invariants: {exc}") from exc
+        finally:
+            if self._auto_commit:
+                session.close()
+
+    @database_retry
+    def remove(self, invariant_id: str) -> bool:
+        """Soft-delete a global/project invariant by setting ``active=False``.
+
+        Args:
+            invariant_id: ULID of the invariant to remove.
+
+        Returns:
+            ``True`` if an invariant was deactivated, ``False`` if not found.
+
+        Raises:
+            DatabaseError: On transient or unexpected DB errors.
+        """
+        session = self._session()
+        try:
+            row = session.query(InvariantModel).filter_by(id=invariant_id).first()
+            if row is None:
+                return False
+            row.active = False
+            session.flush()
+            if self._auto_commit:
+                session.commit()
+                session.close()
+            return True
         except (OperationalError, SQLAlchemyDatabaseError) as exc:
             session.rollback()
             raise DatabaseError(
-                f"Failed to delete correction attempt {correction_attempt_id}: {exc}"
+                f"Failed to remove global invariant {invariant_id}: {exc}",
+            ) from exc
+
+    @database_retry
+    def get_by_id(self, invariant_id: str) -> Any | None:
+        """Look up a single global/project invariant by ID.
+
+        Args:
+            invariant_id: ULID of the invariant.
+
+        Returns:
+            An ``Invariant`` domain object, or ``None``.
+        """
+        session = self._session()
+        try:
+            row = session.query(InvariantModel).filter_by(id=invariant_id).first()
+            if row is None:
+                return None
+            return row.to_domain()
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            raise DatabaseError(
+                f"Failed to get global invariant {invariant_id}: {exc}",
             ) from exc