fix(security): fix validate_path startswith bypass #7478

Implement strategy registry integration as spec §47561 by adding:

- load_strategies_from_config() in context_strategies.py for TOML-driven
  strategy bootstrapping (register builtins, discover custom plugins via
  module:ClassName, set enabled list)
- Re-exports of StrategyRegistry, StrategyConfig, StrategyRegistryEntry,
  StrategyNotFoundError, StrategyRegistrationError and ContextStrategy from
  strategy_registry and acms_service modules
- __all__ export list and DEFAULT_ENABLED_STRATEGIES constant

Fix StrategyRegistry.validate_registry() to skip resource_types check for
v1 pipeline strategies (context_strategies.py, acms_service.py) whose
StrategyCapabilities dataclass lacks the domain-model resource_types field,
preventing false-positive warnings.  Adds _is_v1_pipeline_caps() helper.

Auto-load strategy configuration from Settings in ACMSPipeline.__init__()
when a Settings object is provided, reading context.strategies config and
registering/ enabling strategies via the plugin loader.

.opencode/agents/grooming-worker.md

@@ -0,0 +1,404 @@
+---
+description: >
+  Grooming worker. Performs a full 11-point quality analysis on a single
+  issue or PR and then exits — fixes incorrect or missing labels, missing
+  milestones, missing closing keywords on PRs, orphan hierarchy, stale
+  duplicates, and so on. Operates entirely through the Forgejo REST API;
+  makes no source-code changes. Uses a fixed model — no tier escalation.
+mode: all
+hidden: false
+temperature: 0.1
+model: "CleverThis-15/Qwen3-6-35B-A3B-GGUF-UD-Q3-K-XL"
+reasoningEffort: "high"
+# All worker type agents use the following color
+color: "#00FF00"
+permission:
+  "glob": allow
+  "grep": allow
+  "doom_loop": deny
+
+  # This agent only needs to call one subagent
+  "question": deny
+
+  # The groomer makes NO code changes — every edit pattern is denied.
+  external_directory:
+    "/tmp/**": deny
+    "/app/**": deny
+  edit:
+    "a**": deny
+    "b**": deny
+    "c**": deny
+    "d**": deny
+    "e**": deny
+    "f**": deny
+    "g**": deny
+    "h**": deny
+    "i**": deny
+    "j**": deny
+    "k**": deny
+    "l**": deny
+    "m**": deny
+    "n**": deny
+    "o**": deny
+    "p**": deny
+    "q**": deny
+    "r**": deny
+    "s**": deny
+    "t**": deny
+    "u**": deny
+    "v**": deny
+    "w**": deny
+    "x**": deny
+    "y**": deny
+    "z**": deny
+    "A**": deny
+    "B**": deny
+    "C**": deny
+    "D**": deny
+    "E**": deny
+    "F**": deny
+    "G**": deny
+    "H**": deny
+    "I**": deny
+    "J**": deny
+    "K**": deny
+    "L**": deny
+    "M**": deny
+    "N**": deny
+    "O**": deny
+    "P**": deny
+    "Q**": deny
+    "R**": deny
+    "S**": deny
+    "T**": deny
+    "U**": deny
+    "V**": deny
+    "W**": deny
+    "X**": deny
+    "Y**": deny
+    "Z**": deny
+    "1**": deny
+    "2**": deny
+    "3**": deny
+    "4**": deny
+    "5**": deny
+    "6**": deny
+    "7**": deny
+    "8**": deny
+    "9**": deny
+    "0**": deny
+    "/app/**": deny
+    "/tmp/**": deny
+  read:
+    "**": allow
+
+  # I don't think MCP permissions work, but just in case they do these two should be the only ones usually allowed
+  "sequential-thinking*": allow
+  "context7*": deny
+
+  # The groomer needs Forgejo API access via webfetch; no code search needed.
+  webfetch: allow
+  websearch: deny
+  codesearch: deny
+
+  bash:
+    # All agents should start with deny and then add in as needed
+    "*": deny
+    "echo *": allow
+    "cat *": allow
+    "printenv *": allow
+    "git -C * remote get-url origin": allow
+    "git remote get-url origin": allow
+
+    # The groomer uses curl for Forgejo REST calls and jq for parsing.
+    "jq *": allow
+    "curl *": allow
+
+    # The following bash permissions must be applied to all agents in the auto-agents-system
+    # Block ALL commands that could hit the label creation endpoints
+    "*api/v1/orgs/*/labels*": deny
+    "*api/v1/repos/*/labels*": deny
+    "*https://git.cleverthis.com/api/v1/repos/cleveragents/cleveragents-core/labels*": deny
+
+    "sudo *": deny
+    # CRITICAL: No direct HTTP calls to the OpenCode server
+    "curl*localhost:4096*": deny
+    "curl*127.0.0.1:4096*": deny
+
+    "*force_merge*": deny
+    "*sudo*": deny
+
+  # All the subagents you want this agent to have access to
+  task:
+    # All agents should start with deny and only enable what you need
+    "*": deny
+
+  # All the skills this agent should have access to load
+  skill:
+    # Always start with deny and enable what the agent needs
+    "*": deny
+
+    "cleverthis-guidelines": allow
+---
+
+# Grooming Worker
+
+You perform ONE full quality-analysis pass on ONE issue or ONE pull request and then exit. You never loop, never sleep, and never look for more work.
+
+You **NEVER** modify source code. All of your changes are to Forgejo metadata (labels, milestone, description, dependency links, state transitions, and comments). If grooming a work item would require changes to the source code, you note that in your `[GROOMED]` summary and exit — code changes are someone else's job.
+
+**Note:** This agent uses a fixed model. There is no tier escalation — every grooming pass runs at the same model tier regardless of complexity.
+
+## Behavior
+
+Follow the instructions below exactly as is, no interpretation or modification, you must perform these steps **exactly** how they are described.
+
+### Startup
+
+If you are in a new session, and have not yet initiated startup, then do the following as the very first thing you do. **Never** proceed further until these startup steps are completed.
+
+Startup steps:
+
+1. Parse and validate prompt parameters
+2. Track which variables were **explicitly present** in your prompt vs **fetched** from environment variables or git remote. Only variables explicitly present in your prompt may be passed onward to subagents. Fetched variables must never be propagated through prompts — subagents will fetch them themselves.
+3. Load the `cleverthis-guidelines` skill — its CONTRIBUTING.md content defines the label taxonomy, ticket lifecycle, issue/PR format requirements, and merge checklist that drive every quality check below.
+4. If any required parameters are missing or malformed, exit immediately and report the error
+
+### Main task
+
+Choose the appropriate procedure based on `work_type`. Both procedures share the same Step 2 quality-analysis checklist.
+
+#### Procedure: `issue_groom` (Issue Grooming)
+
+1. **Read the issue.** `GET {forgejo_url}/api/v1/repos/{forgejo_owner}/{forgejo_repo}/issues/{work_number}` — title, body, labels, milestone, assignees, state.
+2. **Paginate all comments.** `GET {forgejo_url}/api/v1/repos/{forgejo_owner}/{forgejo_repo}/issues/{work_number}/comments?limit=50&page=N` — paginate fully.
+3. **Read the issue's labels.** `GET {forgejo_url}/api/v1/repos/{forgejo_owner}/{forgejo_repo}/issues/{work_number}/labels`.
+4. **Read the issue's dependencies.** `GET {forgejo_url}/api/v1/repos/{forgejo_owner}/{forgejo_repo}/issues/{work_number}/dependencies?limit=50&page=N` — paginate fully.
+5. **Run all applicable checks from the Quality Analysis Checklist below.**
+6. **Apply each correction via the Forgejo REST API** (`PATCH .../issues/{work_number}` to edit fields, `POST .../labels` to add labels, `DELETE .../labels/{name}` to remove labels, `POST .../comments` to comment, `POST .../dependencies` to add dependency links).
+7. **Post the `[GROOMED]` marker comment** (see "[GROOMED] Marker" section).
+8. **Exit.**
+
+#### Procedure: `pr_groom` (PR Grooming)
+
+1. **Read the PR.** `GET {forgejo_url}/api/v1/repos/{forgejo_owner}/{forgejo_repo}/pulls/{work_number}` — title, body, labels, milestone, head branch, head SHA, base branch, merged state, mergeable state.
+2. **Paginate all PR comments.** PRs share the issue comment API: `GET {forgejo_url}/api/v1/repos/{forgejo_owner}/{forgejo_repo}/issues/{work_number}/comments?limit=50&page=N` — paginate fully.
+3. **Paginate all formal reviews.** `GET {forgejo_url}/api/v1/repos/{forgejo_owner}/{forgejo_repo}/pulls/{work_number}/reviews?limit=50&page=N` — paginate fully.
+4. **For each review with state `REQUEST_CHANGES`,** `GET {forgejo_url}/api/v1/repos/{forgejo_owner}/{forgejo_repo}/pulls/{work_number}/reviews/{review_id}/comments?limit=50&page=N` — paginate fully and read the inline comments.
+5. **Identify the linked issue.** Parse closing keywords from the PR body (e.g. `Closes #N`, `Fixes #N`, `Resolves #N`). If found, fetch that issue and its labels.
+6. **Run all applicable checks from the Quality Analysis Checklist below** — for PRs, this additionally includes label-sync from the linked issue, closing-keyword presence, and addressing non-code review remarks.
+7. **Apply each correction via the Forgejo REST API.**
+8. **Post the `[GROOMED]` marker comment.**
+9. **Exit.**
+
+### Quality Analysis Checklist
+
+Run every check that applies to the work item type (issue vs PR). Apply each fix via the Forgejo REST API as it is discovered.
+
+#### 1. Duplicate Detection
+Check whether this issue/PR describes the same work as another open item. Search by title and key phrases from the body. If a duplicate is found, close the less-complete one with a comment linking to the more-complete one. Use `PATCH .../issues/{N}` with `state: "closed"` to close.
+
+#### 2. Orphaned Hierarchy
+- For regular issues: verify a parent Epic link exists (this issue should BLOCK its parent Epic).
+- For Epics: verify a parent Legendary link exists.
+- If a parent is missing, add the dependency link via `POST .../issues/{work_number}/dependencies` (or post a comment flagging the orphan if the parent cannot be inferred).
+
+#### 3. Stale Activity Detection
+If the item is in `State/In Progress` and has had **no** activity (comments, commits, label changes) for more than 7 days, post a comment asking whether work is still active. Do **not** silently change state — leave that to a human or a follow-up cycle.
+
+#### 4. Missing Labels
+Every issue and PR must carry a `State/*`, a `Type/*`, and a `Priority/*` label. If any are missing, apply the inferred correct value. Inferences:
+- Closed item with no `State/*` → `State/Completed` (if merged) or `State/Wont Do` (otherwise).
+- Open item with active development → `State/In Progress`.
+- Open item with no PR and no commits yet → leave the state as-is (likely `State/Triaged`).
+
+Apply labels via `POST {forgejo_url}/api/v1/repos/{forgejo_owner}/{forgejo_repo}/issues/{work_number}/labels` with body `{"labels": [<id>, ...]}`. **CRITICAL: never create new labels** — only apply labels that already exist in the repository or org.
+
+#### 5. Incorrect Labels
+Check for contradictions and correct them:
+- Closed issue without `State/Completed` or `State/Wont Do` → set the appropriate state label.
+- Issue with merged PR that is not `State/Completed` → set `State/Completed`.
+- Issue in `State/In Review` without an open PR → revert to `State/In Progress`.
+- Issue in `State/Paused` without `Blocked` label → either add `Blocked` (if blockers exist) or unpause.
+
+To remove a label: `DELETE {forgejo_url}/api/v1/repos/{forgejo_owner}/{forgejo_repo}/issues/{work_number}/labels/{label_id}`.
+
+#### 6. Missing Milestone
+Every issue and PR must have a milestone. If missing, list the repository's open milestones (`GET {forgejo_url}/api/v1/repos/{forgejo_owner}/{forgejo_repo}/milestones?state=open&limit=50&page=N`, paginate fully) and pick the one whose description and due date best match this work item. Apply via `PATCH .../issues/{work_number}` with `milestone: <id>`.
+
+#### 7. Completed Work Not Closed
+If the issue has a linked PR that has been merged but the issue itself is still open, close it (`PATCH .../issues/{work_number}` with `state: "closed"`) and ensure `State/Completed` is applied.
+
+#### 8. Epic/Legendary Completeness
+If this is an Epic, scan its description for scope items. For each scope item that has no corresponding child issue, create one via `POST {forgejo_url}/api/v1/repos/{forgejo_owner}/{forgejo_repo}/issues` (title taken from the scope item, body referencing the Epic, labels matching the Epic's `Priority/*` and `Type/*` from the description's context). Then link the new child as a blocker of the Epic.
+
+#### 9. Dual Status Cleanup (Automation Tracking only)
+If this is an `Automation Tracking` issue (title of the form `[AUTO-*] Status: ...`), find all open tracking issues with the same `[AUTO-*]` prefix (e.g. `[AUTO-IMP-SUP]`). If more than one exists, close all but the newest (by `created_at`).
+
+#### 10. PR-Specific: Label & Milestone Sync With Linked Issue
+For PRs only. Copy the following from the linked issue down to the PR:
+- The `Priority/*` label
+- The `Type/*` label
+- The `MoSCoW/*` label, if present
+- The milestone assignment
+
+If the PR body lacks a closing keyword (`Closes #N`, `Fixes #N`, or `Resolves #N`) referencing the linked issue, edit the PR body to add one: `PATCH {forgejo_url}/api/v1/repos/{forgejo_owner}/{forgejo_repo}/pulls/{work_number}` with the updated `body`.
+
+If the PR is missing a dependency link to the linked issue (PR blocks issue), add one.
+
+If the PR has been merged or closed, ensure both the PR and its linked issue carry `State/Completed`.
+
+#### 11. PR-Specific: Address Non-Code Review Remarks
+Re-read every formal review and review comment. For any concern raised that is **not** about the source code itself — e.g. labels, milestone, PR description, missing closing keyword, MoSCoW classification — address it directly here. Any concern that **is** about source code is left untouched (the implementation worker handles those).
+
+### [GROOMED] Marker
+
+After completing the analysis and all fixes, post a single summary comment on the issue or PR. The comment **must** begin with the literal token `[GROOMED]` so other agents and the orchestrator can detect that this item has been processed. The comment should list every check performed and every fix applied, in the structure shown below.
+
+Comment template:
+```
+[GROOMED] Quality analysis complete.
+
+Checks performed:
+- Duplicate detection: <result>
+- Hierarchy: <result>
+- Activity / staleness: <result>
+- Labels (State / Type / Priority): <result>
+- Label contradictions: <result>
+- Milestone: <result>
+- Closure consistency: <result>
+- Epic completeness: <result>
+- Tracking cleanup: <result>
+- PR label sync with linked issue: <result or n/a>
+- Non-code review remarks: <result or n/a>
+
+Fixes applied:
+- <list each fix, or "none" if nothing changed>
+
+Notes:
+- <Any code-change recommendations that the implementor (not the groomer) should action; or "none".>
+
+---
+Automated by CleverAgents Bot
+Supervisor: Grooming | Agent: grooming-worker
+```
+
+Post via: `POST {forgejo_url}/api/v1/repos/{forgejo_owner}/{forgejo_repo}/issues/{work_number}/comments` with body `{"body": "..."}` and `Authorization: token {forgejo_pat}`.
+
+## Parameters and local variables
+
+Throughout this prompt we will use a format where we will use the local variable name in curly brackets anywhere we want to substitute the contents of that variable. For example, if `{forgejo_owner}` has the value `cleveragents` then `{forgejo_owner}` should be replaced with `cleveragents` wherever it appears.
+
+The following represents all variables this agent works with:
+
+| Parameter           | Local Variable    | Notes                                                                  |
+|---------------------|:-----------------:|------------------------------------------------------------------------|
+| Repository base url | `forgejo_url`     | Base URL for Forgejo API                                               |
+| Repository owner    | `forgejo_owner`   | May be an organization or an individual                                |
+| Repository name     | `forgejo_repo`    | Name of the repository                                                 |
+| Forgejo PAT         | `forgejo_pat`     | Personal access token                                                  |
+| Git name            | `git_user_name`   | Git author name (used in the bot signature)                            |
+| Git email           | `git_user_email`  | Git author email (used in the bot signature)                           |
+| Work type           | `work_type`       | "issue_groom" or "pr_groom"                                            |
+| Work number         | `work_number`     | Issue or PR number                                                     |
+| Work title          | `work_title`      | Title (informational context)                                          |
+
+**CRITICAL:** Parameters given explicitly in the prompt always take precedence. Any value not provided may be resolved through environment variable fallbacks described below.
+
+**CRITICAL — Explicit vs Fetched Variables:** Any value you had to fetch from environment variables or git remote must never be propagated through prompts; subagents are capable of fetching missing variables themselves using their own fallback mechanisms. This applies to **all** variables, both credentials and non-credentials alike.
+
+### What you receive in your prompt
+
+All of the variables listed in the table below may be passed in your prompt. Some are required and some are optional. If a required parameter is missing or malformed you must exit immediately and report the error. Optional parameters that are absent from the prompt can be resolved through fallback mechanisms described in the sections below.
+
+| Parameter           | Required? | Local Variable    |
+|---------------------|:---------:|-------------------|
+| Repository base url | yes       | `forgejo_url`     |
+| Repository owner    | yes       | `forgejo_owner`   |
+| Repository name     | yes       | `forgejo_repo`    |
+| Forgejo PAT         | yes       | `forgejo_pat`     |
+| Git name            | yes       | `git_user_name`   |
+| Git email           | yes       | `git_user_email`  |
+| Work type           | yes       | `work_type`       |
+| Work number         | yes       | `work_number`     |
+| Work title          | yes       | `work_title`      |
+
+Your prompt may also contain parameters beyond those listed in the table above. The orchestrator does not interpret or validate these — they are treated as opaque pass-through values. This agent does not invoke subagents and therefore does not forward extra context anywhere; you simply ignore unrecognised parameters.
+
+#### Example prompt
+
+The following is an example of what a real prompt passed to this agent might look like; real prompts may vary significantly in structure and wording:
+
+```
+forgejo_url: `https://git.cleverthis.com`
+forgejo_owner: `cleveragents`
+forgejo_repo: `cleveragents-core`
+forgejo_pat: `ghp_exampletoken`
+git_user_name: `HAL9000`
+git_user_email: `hal9000@cleverthis.com`
+work_type: `pr_groom`
+work_number: 11197
+work_title: "Fix race condition in async scheduler"
+
+Groom the indicated issue or pull request — fix labels, milestone, description, and other metadata quality problems. Do not make any code changes.
+```
+
+### Variables to fetch
+
+Some optional variables can be auto-detected from the repository context. Only attempt to fetch a variable this way if it was neither provided in the prompt nor found in the corresponding environment variable. The environment variable always takes precedence over the auto-detected value.
+
+| Variable        | Environment Variable | Env var takes precedence? |
+|-----------------|----------------------|:-------------------------:|
+| `forgejo_url`   | `FORGEJO_URL`        | yes                       |
+| `forgejo_owner` | `FORGEJO_OWNER`      | yes                       |
+| `forgejo_repo`  | `FORGEJO_REPO`       | yes                       |
+
+The following are the variables and the steps to fetch them:
+
+- **`forgejo_url`**
+  1. Run `bash("git remote get-url origin")`
+  2. Extract the scheme and host from the output (e.g. `https://git.cleverthis.com`)
+
+- **`forgejo_owner`**
+  1. Run `bash("git remote get-url origin")`
+  2. Parse the first path segment from the URL path
+
+- **`forgejo_repo`**
+  1. Run `bash("git remote get-url origin")`
+  2. Parse the second path segment from the URL path
+  3. Strip any trailing `.git` suffix
+
+### Fallback to environment variables
+
+For optional parameters not provided in your prompt, you may fall back to the environment variables listed below. Always give precedence to values explicitly passed in the prompt. If you attempt to read a required environment variable and it does not exist, exit immediately and report the error.
+
+| Information         | Env Variable      | Required? | Local Variable    |
+|---------------------|-------------------|:---------:|-------------------|
+| Git name            | `GIT_USER_NAME`   | Yes       | `git_user_name`   |
+| Git email           | `GIT_USER_EMAIL`  | Yes       | `git_user_email`  |
+| Forgejo PAT         | `FORGEJO_PAT`     | Yes       | `forgejo_pat`     |
+| Repository base url | `FORGEJO_URL`     | No        | `forgejo_url`     |
+| Repository owner    | `FORGEJO_OWNER`   | No        | `forgejo_owner`   |
+| Repository name     | `FORGEJO_REPO`    | No        | `forgejo_repo`    |
+
+**Note:** The `Required?` column above indicates whether the environment variable must exist if you attempt to use it as a fallback. If you query a required environment variable and it is not set, exit immediately and report the error.
+
+## Subagents
+
+This agent does not invoke any subagents. All grooming actions are performed by direct `curl` calls to the Forgejo REST API (authenticated with the `forgejo_pat`), parsed and composed using `jq` when needed.
+
+## **CRITICAL** Rules
+
+1. **One task, then exit.** Do not loop, do not sleep, do not look for more work.
+2. **Never modify source code.** This agent's job is metadata grooming only. If a quality issue requires a source-code change, note it in the `Notes:` section of the `[GROOMED]` summary and exit — the implementation worker will pick it up. The `edit` permission is set to `deny` across all file paths to enforce this.
+3. **Never create new labels.** All label operations must reference labels that already exist on the repository or org. If a needed label does not exist, note it in the `Notes:` section of the `[GROOMED]` summary and skip that specific fix.
+4. **PR labels and milestone sync from the linked issue.** Priority, Type, MoSCoW, and milestone always flow from the issue to the PR, never the reverse. If the linked issue itself is missing one of those values, fix the issue first, then the PR.
+5. **Always post the `[GROOMED]` marker comment.** Even if no fixes were needed, post a `[GROOMED]` comment listing the checks performed. This is how the orchestrator knows the item has been processed.
+6. **Bot signature on all Forgejo content:**
+   ```
+   ---
+   Automated by CleverAgents Bot
+   Supervisor: Grooming | Agent: grooming-worker
+   ```
+7. **CRITICAL:** Never under **any** circumstances are you to ask any questions of the user. If you have a question, use your best judgement and answer it yourself. Even if you are completely unsure of the answer, make your best guest. It is **COMPLETELY FORBIDDEN** for you to ever ask a question.
+8. **Exhaustive pagination for all list results.** Every REST call returning a list must be paginated fully with `limit=50`. After each response, if the count equals the page size, fetch the next page. Never assume the first response is complete. *Examples specific to this agent:* issue/PR comments (paginate ALL pages — earlier comments contain hierarchy and prior-grooming context); PR reviews and review comments (paginate to read every round of feedback before deciding which remarks need address); repo milestones (paginate when picking the best milestone for an item); issue dependencies (paginate to fully understand the hierarchy).

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

@@ -113,8 +113,8 @@ permission:
     "git remote get-url origin": allow
 
     # This is where we edit the permissions on an as-needed per-agent basis
-    "git -C /tmp/*": allow
-    "cat *": allow
+    # "git -C /tmp/*": allow
+    # "cat *": allow
     "ls *": allow
     "find *": allow
     "grep *": allow

.opencode/agents/task-implementor.md

@@ -151,6 +151,12 @@ permission:
     "git-isolator-util": allow
     "git-commit-util": allow
 
+    # Delegation target for work items that turn out to be pure metadata
+    # quality problems (incorrect labels, missing milestone, etc.) rather
+    # than source-code work. See the "Delegate-or-implement check" subsection
+    # in the agent's main task instructions.
+    "grooming-worker": allow
+
   # All the skills this agent should have access to load
   skill:
     # Always start with deny and enable what the agent needs
@@ -181,7 +187,23 @@ Startup steps:
 
 ### Main task
 
-This is where actual implementation happens. Choose the appropriate procedure based on `work_type` from the subsections below:
+This is where actual implementation happens. Choose the appropriate procedure based on `work_type` from the subsections below.
+
+**Before** entering either procedure, perform the **Delegate-or-implement check** described next.
+
+#### Delegate-or-implement check (run first, every time)
+
+After reading the work item (step 1 of the procedure for `issue_impl`, or steps 1–4 of the procedure for `pr_fix`), but **before** cloning the repository or making any code changes, classify the work item into one of two categories:
+
+- **Code work** — the item requires source-code changes, test changes, configuration changes, infrastructure changes, or any change that lives inside the repository's tracked files. Examples: implementing a feature, fixing a failing test, repairing a CI failure, addressing a `REQUEST_CHANGES` review whose remarks reference specific source files.
+
+- **Metadata-only work** — the item requires only changes to Forgejo metadata that live outside the tracked source tree. Examples: missing/incorrect labels, missing milestone, missing closing keyword in a PR description, missing dependency link to a parent Epic, an issue that should be closed because its PR was merged, a stale `[AUTO-*] Status:` duplicate that should be closed, or PR review remarks whose entire concern is about labels/milestone/description (no code mentioned).
+
+If the item is **metadata-only**, you must **not** implement code. Instead delegate to `grooming-worker` (see the "`grooming-worker`" subsection in the "Subagents" section) and exit. Post no attempt comment of your own — `grooming-worker` will post the `[GROOMED]` marker comment.
+
+If the item is **code work** (even if it also has some metadata problems), proceed with the normal procedure below. Do not split the work. The downstream grooming pool will pick up any leftover metadata corrections in its own pass.
+
+When in doubt, classify as **code work** and proceed normally — the cost of an unnecessary code-work attempt is much lower than the cost of incorrectly skipping legitimate implementation.
 
 #### Procedure: `issue_impl` (New Issue Implementation)
 
@@ -549,13 +571,62 @@ Commit all staged changes and force-push with lease.
 | Repository owner     | `forgejo_owner`  | Passed as context                                              |
 | Repository name      | `forgejo_repo`   | Passed as context                                              |
 
+### `grooming-worker`
+
+#### How to invoke
+
+Invoke `grooming-worker` as a blocking call via the Task tool. Use it **only** when the Delegate-or-implement check in the main-task section classifies the work item as metadata-only. After `grooming-worker` returns, exit immediately — do not also post your own attempt comment, do not clone, do not make any code changes.
+
+`grooming-worker` is a single-tier worker (no tier escalation, no inner task-* agent). You pass it a **flat** prompt — there is no outer/inner structure, and there is no `escalation_tier`. The `work_type` is rewritten from its implementation form to its grooming form:
+
+- `issue_impl` → `issue_groom`
+- `pr_fix` → `pr_groom`
+
+All credentials, `work_number`, and `work_title` pass through unchanged.
+
+#### Prompt template
+
+**Only include a variable line if that variable was explicitly present in your prompt.** Omit any variable you fetched from environment variables — the subagent will fetch it itself.
+
+```
+forgejo_url: `{forgejo_url}`
+forgejo_owner: `{forgejo_owner}`
+forgejo_repo: `{forgejo_repo}`
+forgejo_pat: `{forgejo_pat}`
+git_user_name: `{git_user_name}`
+git_user_email: `{git_user_email}`
+work_type: `{groom_work_type}`
+work_number: `{work_number}`
+work_title: `{work_title}`
+
+Groom the indicated issue or pull request — fix labels, milestone, description, and other metadata quality problems. Do not make any code changes.
+```
+
+Where `{groom_work_type}` is `issue_groom` if your `work_type` was `issue_impl`, or `pr_groom` if your `work_type` was `pr_fix`.
+
+#### Parameters to pass
+
+| Subagent parameter   | Local variable   | Notes                                                                                     |
+|----------------------|:----------------:|-------------------------------------------------------------------------------------------|
+| Repository base url  | `forgejo_url`    | Pass-through                                                                              |
+| Repository owner     | `forgejo_owner`  | Pass-through                                                                              |
+| Repository name      | `forgejo_repo`   | Pass-through                                                                              |
+| Forgejo PAT          | `forgejo_pat`    | Pass-through                                                                              |
+| Git name             | `git_user_name`  | Pass-through (used in the bot signature on the `[GROOMED]` comment)                       |
+| Git email            | `git_user_email` | Pass-through                                                                              |
+| Work type            | derived          | `issue_groom` if your `work_type` was `issue_impl`; `pr_groom` if it was `pr_fix`         |
+| Work number          | `work_number`    | Pass-through                                                                              |
+| Work title           | `work_title`     | Pass-through                                                                              |
+
+After `grooming-worker` returns, return its output verbatim to your caller and exit.
+
 ## **CRITICAL** Rules
 
 1. **One task, then exit.** Do not loop, do not sleep, do not look for more work.
 2. **Never dispatch.** Tier resolution and dispatch happen upstream in `tier-dispatcher`. You are an inner `task-*` agent — do not call `estimator-*`, do not call `tier-*`, and never try to escalate or re-dispatch the work yourself.
 3. **Follow CONTRIBUTING.md exactly.** Commit format, file organisation, testing philosophy, PR requirements — all must be followed. Load the `cleverthis-guidelines` skill for the full CONTRIBUTING.md rules.
 4. **All commands through nox.** Never run `pip install`, `pytest`, `behave`, or `robot` directly.
-5. **Leave an attempt comment always.** Whether you succeeded or failed, post the structured attempt comment. This is how the supervisor tracks escalation state.
+5. **Leave an attempt comment always.** Whether you succeeded or failed, post the structured attempt comment. This is how the supervisor tracks escalation state. The one exception is when you delegate to `grooming-worker` (see Rule 13): in that case the groomer posts a `[GROOMED]` comment and you post nothing additional.
 6. **Never merge.** Create PRs; the merge supervisor handles merging. Never call any merge endpoint.
 7. **Clean up your clone.** Delete the temporary directory before exiting (`rm -rf {repo_dir}`).
 8. **Never work in `/app`.** Always work in `/tmp/`. If `repo_dir` is not inside `/tmp/`, refuse and report an error.
@@ -568,3 +639,4 @@ Commit all staged changes and force-push with lease.
 10. **CRITICAL:** Never under **any** circumstances are you to ask any questions of the user. If you have a question, use your best judgement and answer it yourself. Even if you are completely unsure of the answer, make your best guest. It is **COMPLETELY FORBIDDEN** for you to ever ask a question.
 11. **Exhaustive pagination for all list results.** Every REST call returning a list must be paginated fully with `limit=50`. After each response, if the count equals the page size, fetch the next page. Never assume the first response is complete. *Examples specific to this agent:* issue comments (escalation history may span many pages — missing any change to the tier or attempt history); PR reviews and review comments (paginate to read all feedback rounds before beginning fixes); CI statuses (paginate to find all failing checks).
 12. Never **under any circumstances** ever declare a failing test or issue is blocking. If your CI is failing, regardless of it is pre-existing or any other excuse you **must** fix it, as that is the only way to complete your task. You can create it as a seperate PR, and indicate a blocking dependency, or you can fix it in its own commit. But you **must** fix all CI errors or fix any issues blocking the merge of a PR or ticket and never use the excuse it is out of scope or pre-existing.
+13. **Delegate metadata-only work to `grooming-worker`.** Run the Delegate-or-implement check **before** cloning the repository. If the work item is purely metadata (incorrect labels, missing milestone, missing closing keyword, etc. — no source-code change is required), invoke `grooming-worker` with the appropriate `work_type` (`issue_groom` or `pr_groom`), return its output verbatim, and exit. Do not clone, do not run quality gates, do not post your own attempt comment. The complement of Rule 12 still applies: if there is **any** code-related work involved (a failing CI check, a `REQUEST_CHANGES` review remarking on source code, anything that touches tracked files), classify as code work and proceed normally.

scripts/opencode-builder.sh

@@ -30,6 +30,9 @@ total_max_workers="${CA_MAX_PARALLEL_WORKERS:-4}"
 merge_max_workers=$(awk "BEGIN {print int((${total_max_workers} + 7) / 8)}")
 imp_max_workers="${total_max_workers}"
 rev_max_workers=$(awk "BEGIN {print int((${total_max_workers} + 1) / 2)}")
+# Grooming runs cheap metadata-only work via Forgejo API. Each task is fast
+# (a few REST calls), so we don't need many parallel slots — ceil(N/4).
+groom_max_workers=$(awk "BEGIN {print int((${total_max_workers} + 3) / 4)}")
 
 # Health check intervals (seconds)
 DOOM_CHECK_INTERVAL=900   # 15 minutes
@@ -41,6 +44,27 @@ IDLE_THRESHOLD_MS=180000  # 3 minutes — session considered idle if last_active
 
 # Script paths
 SCRIPT_DIR="/app/.opencode/skills/auto-agents-system/scripts"
+
+# Make tsx discoverable on PATH. Running `npx --yes tsx` concurrently with
+# a cold or partially-warm npm cache causes a race in `~/.npm/_npx/<hash>/`:
+# multiple npm processes simultaneously try to rename `node_modules/esbuild`
+# to the same atomic-backup name and fail with
+# `ENOTEMPTY: rename '.../esbuild' -> '.../esbuild-7kCFO8Lm'`, crashing
+# all but one of them and leaving the cache broken for follow-on calls.
+# When tsx is on PATH, npx skips its install/cache logic entirely and just
+# execs the existing binary — so the race disappears without changing any
+# call sites. We install tsx once into ~/.local/node_modules and prepend
+# its bin dir to PATH for the rest of this script's process tree.
+TSX_PREFIX="${HOME}/.local"
+TSX_BIN_DIR="${TSX_PREFIX}/node_modules/.bin"
+if ! [[ -x "${TSX_BIN_DIR}/tsx" ]]; then
+  printf "[%s] Installing tsx into %s …\n" "$(date +%H:%M:%S)" "$TSX_PREFIX" >&2
+  mkdir -p "$TSX_PREFIX"
+  npm install --prefix "$TSX_PREFIX" --silent tsx >/dev/null 2>&1 \
+    || { printf "ERROR: failed to install tsx into %s\n" "$TSX_PREFIX" >&2; exit 1; }
+fi
+export PATH="${TSX_BIN_DIR}:${PATH}"
+
 SESSION_LIST="npx --yes tsx ${SCRIPT_DIR}/session_list.ts"
 SESSION_FIND_PREFIX="npx --yes tsx ${SCRIPT_DIR}/session_find_by_prefix.ts"
 SESSION_START="npx --yes tsx ${SCRIPT_DIR}/session_start.ts"
@@ -251,6 +275,14 @@ send_message_capture() {
   local body="$2"
   local out_file="$3"
 
+  # Refuse to send empty bodies — the server logs them as
+  # "No user message found in stream. This should never happen."
+  if [[ -z "$body" ]]; then
+    log_warn "send_message_capture: refusing to send empty body to ${sid}"
+    : > "$out_file" 2>/dev/null || true
+    return 1
+  fi
+
   # Use prompt_async — fire-and-forget; server returns 204 immediately.
   # The synchronous /message endpoint would block until the agent finishes
   # its full response (potentially many minutes for a busy worker).
@@ -448,6 +480,27 @@ build_impl_prompt() {
 }'
 }
 
+build_groom_prompt() {
+  local item_json="$1"
+  local kind="$2"
+  local work_num="$3"
+
+  local work_type; work_type=$([ "$kind" == "ISSUE" ] && echo "issue_groom" || echo "pr_groom")
+  local compact_item
+  compact_item=$(echo "$item_json" | jq -c .)
+
+  jq -n \
+    --arg wn "$work_num" \
+    --arg wt "$work_type" \
+    --argjson ij "$compact_item" \
+    '{
+  issue_number: $wn,
+  pr_number: $wn,
+  work_type: $wt,
+  item_json: $ij
+}'
+}
+
 build_merge_prompt() {
   local pr_json="$1"
   local pr_num="$2"
@@ -616,6 +669,13 @@ Evaluate the health of this session."
   printf '%s' "$eval_prompt" > "$eval_tmp"
   local eval_body
   eval_body=$(jq -nc --arg a "worker-health-evaluator" --rawfile t "$eval_tmp" '{agent: $a, parts: [{type: "text", text: $t}]}')
+  if [[ -z "$eval_body" ]]; then
+    # jq failed (likely couldn't read eval_tmp) — bail rather than send empty
+    rm -f "$eval_tmp"
+    delete_session "$eval_sid"
+    echo "ok"
+    return
+  fi
   curl -sf -X POST "${BASE}/session/${eval_sid}/prompt_async" -H "Content-Type: application/json" -d "$eval_body" >/dev/null 2>&1
   local eval_deadline=$(( $(date +%s) + 120 ))
   while [[ $(date +%s) -lt $eval_deadline ]]; do
@@ -671,7 +731,7 @@ run_health_cycle() {
 
   # 2) Idle worker check — collect all idle SIDs across prefixes, then fetch
   # their messages in parallel before processing each one.
-  local all_prefixes=("AUTO-IMP" "AUTO-MRG" "AUTO-REV")
+  local all_prefixes=("AUTO-IMP" "AUTO-MRG" "AUTO-REV" "AUTO-GROOM")
   local -a idle_sids=()
   for prefix in "${all_prefixes[@]}"; do
     local sessions_json
@@ -850,7 +910,7 @@ fi
 # ── Main orchestration loop ───────────────────────────────────────────────
 main_loop() {
   log "Starting orchestration loop."
-  log "Worker pools: merge=${merge_max_workers} imp=${imp_max_workers} rev=${rev_max_workers}"
+  log "Worker pools: merge=${merge_max_workers} imp=${imp_max_workers} rev=${rev_max_workers} groom=${groom_max_workers}"
 
   while ! $STOP_LOOP; do
     N=$((N + 1))
@@ -913,11 +973,28 @@ main_loop() {
         "Changes Requested|${SCRIPT_DIR}/list_prs_changes_requested.ts|PR"
         "CI Failing|${SCRIPT_DIR}/list_prs_ci_failing.ts|PR"
       )
+      # GROOM mirrors IMP: same priority order (PRs first, issues second), same
+      # work groups. Re-uses IMP's cached results within the same iteration
+      # via the shared cache_dir, so grooming pre-fetch is effectively free.
+      declare -a GROOM_WORK_GROUPS=(
+        "CI Failing PRs|${SCRIPT_DIR}/list_prs_ci_failing.ts|PR"
+        "Addressed Changes CI Passing|${SCRIPT_DIR}/list_prs_addressed_changes_ci_passing.ts|PR"
+        "Addressed Changes CI Failing|${SCRIPT_DIR}/list_prs_addressed_changes_ci_failing.ts|PR"
+        "Changes Requested|${SCRIPT_DIR}/list_prs_changes_requested.ts|PR"
+        "CI Missing|${SCRIPT_DIR}/list_prs_missing_ci_checks.ts|PR"
+        "No Active Review CI Passing|${SCRIPT_DIR}/list_prs_no_active_review_ci_passing.ts|PR"
+        "No Active Review CI Failing|${SCRIPT_DIR}/list_prs_no_active_review_ci_failing.ts|PR"
+        "Needs Review Not Stale|${SCRIPT_DIR}/list_prs_needs_review_not_stale.ts|PR"
+        "Needs Review Stale Clean|${SCRIPT_DIR}/list_prs_needs_review_stale_clean.ts|PR"
+        "Needs Review Stale Conflicts|${SCRIPT_DIR}/list_prs_needs_review_stale_conflicts.ts|PR"
+        "Open Issues|${SCRIPT_DIR}/list_issues.ts|ISSUE"
+      )
 
       declare -a pools=(
         "AUTO-IMP|implementation-worker|${imp_max_workers}|IMP_WORK_GROUPS"
         "AUTO-MRG|pr-merge-worker|${merge_max_workers}|MRG_WORK_GROUPS"
         "AUTO-REV|pr-review-worker|${rev_max_workers}|REV_WORK_GROUPS"
+        "AUTO-GROOM|grooming-worker|${groom_max_workers}|GROOM_WORK_GROUPS"
       )
 
       # Shared cache dir for this iteration — pools reuse each other's results
@@ -1029,7 +1106,7 @@ main_loop() {
           elif [[ $ec -ne 0 ]]; then
             log_warn "[pool:${prefix}]   ${ck}: FAILED (exit ${ec}) — treating as empty"
             local snippet
-            snippet=$(head -n 3 "${out_file}.err" 2>/dev/null | tr '\n' '|' | sed 's/|$//')
+            snippet=$(head -n 20 "${out_file}.err" 2>/dev/null | tr '\n' '|' | sed 's/|$//')
             if [[ -n "$snippet" ]]; then
               log_warn "  stderr: ${snippet}"
             fi
@@ -1144,15 +1221,33 @@ main_loop() {
               local tmp_prompt
               tmp_prompt=$(mktemp "/tmp/ocb-${worker_tag}-${$}.XXXXXX.txt") || continue
 
-              local prompt_text
+              local prompt_text=""
               case "$agent" in
                 *implementation*) prompt_text=$(build_impl_prompt "$item" "$kind" "$work_num") ;;
                 *merge*)          prompt_text=$(build_merge_prompt "$item" "$work_num") ;;
                 *review*)         prompt_text=$(build_review_prompt "$item" "$work_num") ;;
+                *grooming*)       prompt_text=$(build_groom_prompt "$item" "$kind" "$work_num") ;;
               esac
 
+              # Skip if the build function returned an empty prompt — sending
+              # an empty body to prompt_async results in "no user message
+              # found in stream" errors on the server.
+              if [[ -z "$prompt_text" ]]; then
+                log_warn "[pool:${prefix}]   built empty prompt for ${kind} ${work_num} (${agent}) — skipping launch"
+                rm -f "$tmp_prompt"
+                continue
+              fi
+
               printf '%s' "$prompt_text" > "$tmp_prompt"
 
+              # Belt-and-braces: confirm the file actually has content on disk
+              # before launching the session.
+              if [[ ! -s "$tmp_prompt" ]]; then
+                log_warn "[pool:${prefix}]   prompt file empty after write for ${kind} ${work_num} (${agent}) — skipping launch"
+                rm -f "$tmp_prompt"
+                continue
+              fi
+
               $SESSION_START --server "${BASE}" --tag "$worker_tag" --agent "$agent" --prompt-file "$tmp_prompt" --restart \
                 >/dev/null 2>&1 &
               pids+=($!)

src/cleveragents/application/services/acms_service.py

@@ -759,6 +759,14 @@ class ACMSPipeline:
         # can wire configuration (e.g. default budget, strategy allow-lists)
         # and persistence without changing the constructor signature.
         self._settings = settings
+
+        # Auto-load strategy configuration from Settings if provided.
+        # Per spec §42947-42980: when a Settings object is passed to the
+        # pipeline, its TOML-derived ``context`` configuration is used to
+        # populate and enable the StrategyRegistry.  This eliminates the need
+        # for callers to manually register every strategy.
+        if settings is not None:
+            self._load_strategies_from_settings(settings)
         self._unit_of_work = unit_of_work
 
         self._strategies: dict[str, ContextStrategy] = {
@@ -802,6 +810,57 @@ class ACMSPipeline:
         self._plugin_manager = plugin_manager
         self._discover_context_extensions()
 
+    def _load_strategies_from_settings(self, settings: Settings) -> None:
+        """Auto-load strategy configuration from a :class:`~cleveragents.config.settings.Settings` object.
+
+        Attempts to read ``context.strategies`` configuration from the TOML-based
+        Settings (typically loaded from a project config file via pydantic-settings).
+        If present, built-in strategies are registered as per default, custom plugins
+        from ``module:ClassName`` strings are discovered, and the enabled list is set.
+        Any errors are logged but do not prevent the pipeline from operating with
+        its core built-in strategies.
+        """
+        try:
+            context_cfg = getattr(settings, "context", None)
+            if context_cfg is None:
+                return
+            # Extract the strategies sub-section as a dict.
+            strategy_cfg: dict[str, Any] = {}
+            if isinstance(context_cfg, dict):
+                strat_dict = context_cfg.get("strategies", {}) or {}
+                if isinstance(strat_dict, dict):
+                    strategy_cfg = dict(strat_dict)
+            else:
+                return
+
+            # Import the lazy-load helper; avoid circular import at module level.
+            from cleveragents.application.services.context_strategies import (
+                load_strategies_from_config,
+            )
+
+            registry = load_strategies_from_config(strategy_cfg)
+            for name in registry.list_all():
+                if name not in self._strategies:
+                    strat_inst = registry.get(name)
+                    self.register_strategy(name=name, strategy=strat_inst)
+            # Sync the enabled list.
+            enabled = registry.list_enabled()
+            if enabled:
+                self._enforce_enabled_strategies(enabled)
+
+        except Exception as exc:  # noqa: BLE001
+            self._logger.warning(
+                "strategy_config_load_failed",
+                error=str(exc),
+            )
+
+    def _enforce_enabled_strategies(self, enabled_names: list[str]) -> None:
+        """Internal helper to filter ``_strategies`` to the enabled subset."""
+        enabled_set = set(enabled_names)
+        self._strategies = {
+            name: strat for name, strat in self._strategies.items() if name in enabled_set
+        }
+
     # ------------------------------------------------------------------
     # Plugin extension point discovery
     # ------------------------------------------------------------------

src/cleveragents/application/services/context_strategies.py

@@ -17,6 +17,11 @@ All strategies implement the v1 ``ContextStrategy`` Protocol defined in
 ``acms_service.py`` and can be registered with ``ACMSPipeline`` via
 ``register_strategy()`` or added to ``BUILTIN_STRATEGIES``.
 
+**Registry integration** (spec §47561): This module is the canonical home for
+:py:class:`~cleveragents.application.services.strategy_registry.StrategyRegistry`.
+The strategies here are registered by default when a registry is created, and
+custom strategies can be loaded via :func:`load_strategies_from_config`.
+
 Based on ``docs/specification.md`` §25207-25216.
 """
 
@@ -36,6 +41,41 @@ from cleveragents.domain.models.core.context_fragment import (
     ContextFragment,
 )
 
+# ---------------------------------------------------------------------------
+# Re-exports for registry integration (spec §47561)
+# ---------------------------------------------------------------------------
+
+from cleveragents.application.services.strategy_registry import (
+    StrategyConfig,            # noqa: F401
+    StrategyNotFoundError,     # noqa: F401
+    StrategyRegistrationError, # noqa: F401
+    StrategyRegistry,          # noqa: F401
+    StrategyRegistryEntry,     # noqa: F401
+)
+
+from cleveragents.domain.models.acms.strategy import (
+    ContextStrategy,             # noqa: F401
+    StrategyCapabilities as _DomainStrategyCapabilities,  # noqa: F401
+)
+
+__all__: list[str] = [
+    "SimpleKeywordStrategy",
+    "SemanticEmbeddingStrategy",
+    "BreadthDepthNavigatorStrategy",
+    "StrategyRegistry",
+    "StrategyConfig",
+    "StrategyRegistryEntry",
+    "StrategyNotFoundError",
+    "StrategyRegistrationError",
+]
+
+# Built-in strategy registry constants (used by load_strategies_from_config)
+DEFAULT_ENABLED_STRATEGIES: tuple[str, ...] = (
+    "simple-keyword",
+    "semantic-embedding",
+    "breadth-depth-navigator",
+)
+
 logger = logging.getLogger(__name__)
 
 _WORD_RE = re.compile(r"\w+", re.UNICODE)
@@ -403,3 +443,59 @@ def _max_proximity(node_uri: str, focus_nodes: list[str], max_hops: int) -> floa
             best = max(best, proximity)
 
     return best
+
+# ---------------------------------------------------------------------------
+# Helper: load_strategies_from_config (spec §42947-42980)
+# ---------------------------------------------------------------------------
+
+
+def load_strategies_from_config(config: dict[str, Any]) -> StrategyRegistry:
+    """Populate and return a :class:`StrategyRegistry` from a TOML config dict.
+
+    Accepts a dict representing the parsed ``[context.strategies]`` section of
+    a TOML configuration file.  The expected structure is::
+
+        {
+            "enabled": ["simple-keyword", "semantic-embedding"],
+            "custom": {
+                "my-strategy": "my_package.strategies:MyStrategy",
+            },
+        }
+
+    :param config: Dict with an optional ``"enabled"`` key and an optional
+        ``"custom"`` dict mapping strategy names to ``"module:ClassName"``
+        strings.
+    :returns: A fully-populated :class:`StrategyRegistry` ready for use by
+        an ``ACMSPipeline`` or downstream component.
+
+    The built-in strategies from :const:`DEFAULT_ENABLED_STRATEGIES` are
+    always registered first (as built-ins).  Custom strategies are discovered
+    and loaded via :meth:`StrategyRegistry.register_from_module`.  Finally,
+    the enabled list is set so that only configured strategies are active.
+    """
+    registry = StrategyRegistry()
+
+    # Register all built-in strategies from DEFAULT_ENABLED_STRATEGIES
+    for strategy_class_name in (
+        "SimpleKeywordStrategy",
+        "SemanticEmbeddingStrategy",
+        "BreadthDepthNavigatorStrategy",
+    ):
+        cls = globals()[strategy_class_name]
+        instance = cls()
+        registry.register(
+            instance,
+            config=StrategyConfig(),
+            module_path="cleveragents.application.services.context_strategies",
+            is_builtin=True,
+        )
+
+    # Register custom strategies from module:ClassName strings
+    for name, module_path in (config.get("custom") or {}).items():
+        registry.register_from_module(name=name, module_path=module_path)
+
+    # Set the enabled strategy list
+    enabled = config.get("enabled", DEFAULT_ENABLED_STRATEGIES)
+    registry.set_enabled(list(enabled))
+
+    return registry

src/cleveragents/application/services/decomposition_clustering.py

@@ -19,6 +19,7 @@ from __future__ import annotations
 
 import os
 from collections import defaultdict
+from pathlib import Path
 
 # ---------------------------------------------------------------------------
 # Token estimation
@@ -74,20 +75,22 @@ def _directory_key(path: str, depth: int = 2, root: str | None = None) -> str:
     Returns:
         The directory key (first *depth* components of the path).
     """
-    # Normalize path separators
-    normalized = path.replace("\\", "/")
+    # Normalize to pathlib for safe, path-semantic comparisons — never
+    # string prefix matching (startswith) which is vulnerable to
+    # /tmp/sandboxmalicious prefix-collision bypass — issue #7478.
+    norm_path = Path(path)
+    parts_raw: list[str]
 
-    # If root is provided, compute relative path
     if root:
-        root_normalized = root.replace("\\", "/")
-        # Ensure root ends with / for proper prefix matching
-        if not root_normalized.endswith("/"):
-            root_normalized += "/"
-        # Remove root prefix if path starts with it
-        if normalized.startswith(root_normalized):
-            normalized = normalized[len(root_normalized) :]
-
-    parts = normalized.split("/")
+        norm_root = Path(root)
+        try:
+            relative = str(norm_path.relative_to(norm_root))
+        except ValueError:
+            # path is not under root; use the original path as-is.
+            relative = path
+        parts_raw = relative.replace("\\", "/").split("/")
+    else:
+        parts_raw = norm_path.parts
     # Filter out empty parts (from leading / in absolute paths)
     parts = [p for p in parts if p]
 

src/cleveragents/application/services/llm_actors.py

@@ -10,6 +10,7 @@ from __future__ import annotations
 
 import os
 import re
+from pathlib import Path
 from datetime import UTC, datetime
 from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable
 
@@ -490,9 +491,13 @@ class LLMExecuteActor:
         for match in pattern.finditer(llm_output):
             path = match.group(1).strip()
             content = match.group(2)
-            full_path = os.path.normpath(os.path.join(sandbox_root, path))
-            # Path traversal guard: reject paths escaping sandbox
-            if not full_path.startswith(sandbox_root + os.sep):
+            full_path_str = os.path.normpath(os.path.join(sandbox_root, path))
+            full_path = Path(full_path_str).resolve()
+            sandbox_resolved = Path(sandbox_root).resolve()
+            # Path traversal guard: reject paths escaping sandbox using
+            # path-semantic comparison (never str.startswith) to prevent
+            # /tmp/sandboxmalicious bypass — see issue #7478.
+            if not (full_path == sandbox_resolved or full_path.is_relative_to(sandbox_resolved)):
                 logger.warning(
                     "Rejected path traversal in LLM output",
                     path=path,

src/cleveragents/application/services/strategy_registry.py

@@ -23,6 +23,9 @@ from typing import Any
 
 import structlog
 
+from cleveragents.application.services.acms_service import (
+    StrategyCapabilities as _V1StrategyCapabilities,
+)
 from cleveragents.domain.models.acms.strategy import (
     ContextStrategy,
     StrategyConfig,
@@ -32,6 +35,24 @@ from cleveragents.domain.models.acms.strategy import (
 logger = structlog.get_logger(__name__)
 
 
+def _is_v1_pipeline_caps(caps: object) -> bool:
+    """Return True if *caps* is a v1-pipeline StrategyCapabilities dataclass.
+
+    V1 pipeline strategies (those defined in ``context_strategies.py`` and
+    ``acms_service.py``) use a dataclass-style capabilities object that has
+    attributes like :attr:`~StrategyCapabilities.supports_semantic_search`,
+    rather than the domain-model ``StrategyCapabilities`` (from
+    ``strategy.py``) which has :attr:`~StrategyCapabilities.uses_text`,
+    :attr:`~StrategyCapabilities.resource_types`, etc.
+
+    We detect v1-capabilities by checking whether the object is an instance
+    of the acms_service dataclass.  This allows :meth:`validate_registry` to
+    skip the resource_types check for legacy strategies that do not declare
+    domain-model backend fields.
+    """
+    return isinstance(caps, _V1StrategyCapabilities)
+
+
 # ---------------------------------------------------------------------------
 # Exceptions
 # ---------------------------------------------------------------------------
@@ -489,7 +510,12 @@ class StrategyRegistry:
                         f"Strategy '{name}' declares no backend capabilities"
                     )
 
-                if not caps.resource_types:
+                # Skip the resource_types check for v1 pipeline strategies.
+                # Those use a dataclass-style StrategyCapabilities that lacks
+                # the domain-model ``resource_types`` field entirely, so all
+                # warnings about undeclared resource types would be false
+                # positives.  See _is_v1_pipeline_caps().
+                if not _is_v1_pipeline_caps(caps) and not caps.resource_types:
                     warnings.append(
                         f"Strategy '{name}' does not declare supported "
                         f"resource types (capabilities.resource_types is empty)"

src/cleveragents/cli/commands/plan.py

@@ -979,7 +979,11 @@ def _apply_sandbox_changes(
             src = os.path.join(dirpath, fname)
             rel = os.path.relpath(src, sandbox_root)
             dst = os.path.normpath(os.path.join(project_root, rel))
-            if not dst.startswith(project_root + os.sep):
+            # Path containment via pathlib to avoid prefix-collision bypass
+            # (/tmp/sandboxmalicious) — see issue #7478.
+            _dst_path = Path(dst).resolve()
+            _pr_path = Path(project_root).resolve()
+            if not (_dst_path == _pr_path or _dst_path.is_relative_to(_pr_path)):
                 continue
             if rel.split(os.sep)[0] in _skip_dirs:
                 continue

src/cleveragents/resource/handlers/_base.py

@@ -182,9 +182,7 @@ class BaseResourceHandler:
         """
         root = Path(location).resolve()
         target = (root / path).resolve()
-        # Use root + os.sep to prevent prefix collision bypass:
-        # e.g. root=/tmp/foo must not match target=/tmp/foobar/secret
-        if target != root and not str(target).startswith(str(root) + os.sep):
+        if not (target == root or target.is_relative_to(root)):
             raise PermissionError(f"Path '{path}' escapes resource root '{location}'")
         return target
 

src/cleveragents/skills/builtins/file_ops.py

@@ -73,12 +73,12 @@ def validate_sandbox_path(path_str: str, sandbox_root: str | None = None) -> Pat
     if not path_str:
         raise ValueError("Path must not be empty")
 
-    root = Path(sandbox_root) if sandbox_root else Path.cwd()
-    root = root.resolve()
-
+    root = Path(sandbox_root).resolve() if sandbox_root else Path.cwd().resolve()
     target = (root / path_str).resolve()
-    if not str(target).startswith(str(root)):
-        raise ValueError(f"Path traversal detected: '{path_str}' escapes sandbox root")
+    if not (target == root or target.is_relative_to(root)):
+        raise ValueError(
+            f"Path traversal detected: '{path_str}' escapes sandbox root '{root}'"
+        )
     return target
 
 

src/cleveragents/skills/inline_executor.py

@@ -263,7 +263,7 @@ class InlineToolExecutor:
                 try:
                     resolved = Path(value).resolve()
                     sandbox_resolved = sandbox_path.resolve()
-                    if not str(resolved).startswith(str(sandbox_resolved)):
+                    if not (resolved == sandbox_resolved or resolved.is_relative_to(sandbox_resolved)):
                         return (
                             f"Path '{value}' for key '{key}' escapes sandbox "
                             f"root '{sandbox_path}'"

src/cleveragents/tool/builtins/file_tools.py

@@ -79,16 +79,12 @@ def validate_path(path_str: str, sandbox_root: str | None = None) -> Path:
     Raises ``ValueError`` when the resolved path escapes *sandbox_root*
     (defaults to the current working directory when not supplied).
     """
-    root = Path(sandbox_root) if sandbox_root else Path.cwd()
-    root = root.resolve()
-
+    root = Path(sandbox_root).resolve() if sandbox_root else Path.cwd().resolve()
     target = (root / path_str).resolve()
-    try:
-        target.relative_to(root)
-    except ValueError as exc:
+    if not (target == root or target.is_relative_to(root)):
         raise ValueError(
-            f"Path traversal detected: '{path_str}' escapes sandbox root"
-        ) from exc
+            f"Path traversal detected: '{path_str}' escapes sandbox root '{root}'"
+        )
     return target
 
 

tests/strategies/test_strategy_registry.py

@@ -0,0 +1,786 @@
+"""Tests for the ContextStrategyRegistry and plugin loading system.
+
+Covers:
+- Strategy registration and retrieval
+- Protocol conformance validation
+- Plugin loading from ``module:ClassName`` strings
+- TOML configuration loading via :func:`load_strategies_from_config`
+- Thread safety of registry operations
+- Fallback degradation chain verification
+- Config-driven strategy enable/disable
+"""
+
+from __future__ import annotations
+
+import threading
+import time
+from collections.abc import Sequence
+from typing import Any
+
+import pytest
+
+# ---------------------------------------------------------------------------
+# Test fixtures and helpers
+# ---------------------------------------------------------------------------
+
+
+class FakeStrategy:
+    """Minimal strategy that satisfies the v1 ``ContextStrategy`` protocol."""
+
+    def __init__(self, name: str = "fake") -> None:
+        self._name = name
+
+    @property
+    def name(self) -> str:  # type: ignore[override]
+        return self._name
+
+    @property
+    def capabilities(self):
+        from cleveragents.application.services.acms_service import (
+            StrategyCapabilities as _Caps,
+        )
+        return _Caps(supports_semantic_search=False)
+
+    def can_handle(self, request: dict[str, Any]) -> float:  # type: ignore[override]
+        return 0.5
+
+    def assemble(
+        self,
+        fragments: Sequence[Any],
+        budget: Any,
+    ) -> Sequence[Any]:
+        return list(fragments)
+
+    def explain(self) -> str:
+        return "Fake strategy for testing."
+
+
+class FakeDomainStrategy:
+    """Strategy that implements the domain-model ``ContextStrategy`` Protocol."""
+
+    def __init__(self, name: str = "domain-fake") -> None:
+        self._name = name
+
+    @property
+    def name(self) -> str:  # type: ignore[override]
+        return self._name
+
+    @property
+    def capabilities(self):
+        from cleveragents.domain.models.acms.strategy import StrategyCapabilities
+        return StrategyCapabilities(uses_text=True, resource_types=("text",))
+
+    def can_handle(self, request: dict[str, Any], backends: Any) -> float:  # type: ignore[override]
+        return 0.8
+
+    def assemble(
+        self, request: Any, backends: Any, budget: int, plan_context: Any
+    ) -> list:  # type: ignore[override]
+        return []
+
+    def explain(self) -> str:
+        return "Fake domain strategy for testing."
+
+
+class FakeStrategyWithNoCaps:
+    """Strategy with empty capabilities (used to trigger validate_registry warnings)."""
+
+    @property
+    def name(self) -> str:
+        return "noop-cap-fake"
+
+    @property
+    def capabilities(self):
+        from cleveragents.domain.models.acms.strategy import StrategyCapabilities
+        return StrategyCapabilities(uses_text=False, resource_types=())  # empty caps
+
+    def can_handle(self, request: dict[str, Any], backends: Any) -> float:
+        return 0.1
+
+    def assemble(self, request: Any, backends: Any, budget: int, plan_context: Any) -> list:
+        return []
+
+    def explain(self) -> str:
+        return "Strategy with empty capabilities"
+
+
+# ---------------------------------------------------------------------------
+# Tests: Strategy registration and retrieval
+# ---------------------------------------------------------------------------
+
+
+class TestRegistrationAndRetrieval:
+    """Validate basic register/get/list operations."""
+
+    def test_register_and_get(self):
+        from cleveragents.application.services.strategy_registry import (
+            StrategyRegistry,
+        )
+
+        registry = StrategyRegistry()
+        strategy = FakeStrategy("my-strategy")
+        registry.register(strategy)
+        assert registry.get("my-strategy") is strategy
+        assert "my-strategy" in registry
+
+    def test_register_with_custom_name(self):
+        from cleveragents.application.services.strategy_registry import (
+            StrategyRegistry,
+        )
+
+        registry = StrategyRegistry()
+        strategy = FakeStrategy("original-name")
+        registry.register(strategy, name="override-name")
+        assert "override-name" in registry
+        assert "original-name" not in registry
+
+    def test_register_with_config(self):
+        from cleveragents.application.services.strategy_registry import (
+            StrategyConfig,
+            StrategyRegistry,
+        )
+
+        registry = StrategyRegistry()
+        strategy = FakeStrategy("config-strategy")
+        cfg = StrategyConfig(enabled=False)
+        registry.register(strategy, config=cfg)
+
+        # With enabled=False the strategy should be registered but not in _enabled_order
+        entry = registry.get_entry("config-strategy")
+        assert entry.config.enabled is False
+
+    def test_register_builtin_flag(self):
+        from cleveragents.application.services.strategy_registry import (
+            StrategyRegistry,
+        )
+
+        registry = StrategyRegistry()
+        strategy = FakeStrategy("built-in-1")
+        registry.register(strategy, is_builtin=True)
+
+        assert "built-in-1" in registry.list_builtin()
+        entry = registry.get_entry("built-in-1")
+        assert entry.is_builtin is True
+
+    def test_duplicate_registration_raises(self):
+        from cleveragents.application.services.strategy_registry import (
+            StrategyRegistrationError,
+            StrategyRegistry,
+        )
+
+        registry = StrategyRegistry()
+        s = FakeStrategy("dup")
+        registry.register(s)
+        with pytest.raises(StrategyRegistrationError):
+            registry.register(FakeStrategy("dup"))
+
+    def test_empty_name_raises(self):
+        from cleveragents.application.services.strategy_registry import (
+            StrategyRegistrationError,
+            StrategyRegistry,
+        )
+
+        registry = StrategyRegistry()
+
+        class Nameless:
+            @property
+            def name(self) -> str:  # type: ignore[override]
+                return ""
+
+            @property
+            def capabilities(self):
+                from cleveragents.application.services.acms_service import (
+                    StrategyCapabilities as _Caps,
+                )
+                return _Caps()
+
+            def can_handle(self, request: dict[str, Any]) -> float:  # type: ignore[override]
+                return 0.3
+
+            def assemble(
+                self, fragments: Sequence[Any], budget: Any
+            ) -> Sequence[Any]:
+                return []
+
+            def explain(self) -> str:
+                return ""
+
+        with pytest.raises(StrategyRegistrationError, match="empty"):
+            registry.register(Nameless())
+
+    def test_protocol_violation_raises(self):
+        from cleveragents.application.services.strategy_registry import (
+            StrategyRegistrationError,
+            StrategyRegistry,
+        )
+
+        registry = StrategyRegistry()
+
+        class NotStrategy:  # No name/abilities etc. at all
+            pass
+
+        with pytest.raises(StrategyRegistrationError):
+            registry.register(NotStrategy())  # type: ignore[arg-type]
+
+    def test_list_all(self):
+        from cleveragents.application.services.strategy_registry import (
+            StrategyRegistry,
+        )
+
+        registry = StrategyRegistry()
+        for i in range(5):
+            entry_name = f"strat-{i}"
+            registry.register(FakeStrategy(entry_name))
+        all_names = registry.list_all()
+        assert sorted(all_names) == ["strat-0", "strat-1", "strat-2", "strat-3", "strat-4"]
+
+    def test_list_enabled_empty(self):
+        from cleveragents.application.services.strategy_registry import (
+            StrategyRegistry,
+        )
+
+        registry = StrategyRegistry()
+        assert registry.list_enabled() == []
+
+    def test_get_entry_and_config(self):
+        from cleveragents.application.services.strategy_registry import (
+            StrategyConfig,
+            StrategyNotFoundError,
+            StrategyRegistry,
+        )
+
+        registry = StrategyRegistry()
+        cfg = StrategyConfig(timeout_seconds=60)
+        registry.register(FakeStrategy("cfg-strat"), config=cfg)
+
+        entry = registry.get_entry("cfg-strat")
+        assert entry.name == "cfg-strat"
+        assert entry.config.timeout_seconds == 60
+
+        loaded_cfg = registry.get_config("cfg-strat")
+        assert loaded_cfg.timeout_seconds == 60
+
+    def test_get_raises_on_missing(self):
+        from cleveragents.application.services.strategy_registry import (
+            StrategyNotFoundError,
+            StrategyRegistry,
+        )
+
+        registry = StrategyRegistry()
+        with pytest.raises(StrategyNotFoundError, match="not found"):
+            registry.get("nonexistent")
+
+    def test_len_and_contains(self):
+        from cleveragents.application.services.strategy_registry import (
+            StrategyRegistry,
+        )
+
+        registry = StrategyRegistry()
+        assert len(registry) == 0
+        assert "empty" not in registry
+
+        registry.register(FakeStrategy("one"))
+        assert len(registry) == 1
+        assert "one" in registry
+
+    def test_unregister(self):
+        from cleveragents.application.services.strategy_registry import (
+            StrategyNotFoundError,
+            StrategyRegistry,
+        )
+
+        registry = StrategyRegistry()
+        registry.register(FakeStrategy("to-unregister"))
+        assert "to-unregister" in registry
+
+        registry.unregister("to-unregister")
+        assert "to-unregister" not in registry
+
+    def test_unregister_unknown_raises(self):
+        from cleveragents.application.services.strategy_registry import (
+            StrategyNotFoundError,
+            StrategyRegistry,
+        )
+
+        registry = StrategyRegistry()
+        with pytest.raises(StrategyNotFoundError):
+            registry.unregister("ghost")
+
+    def test_clear(self):
+        from cleveragents.application.services.strategy_registry import (
+            StrategyRegistry,
+        )
+
+        registry = StrategyRegistry()
+        for i in range(3):
+            registry.register(FakeStrategy(f"clear-{i}"))
+        assert len(registry) == 3
+
+        registry.clear()
+        assert len(registry) == 0
+        assert registry.list_all() == []
+
+
+# ---------------------------------------------------------------------------
+# Tests: Protocol conformance validation
+# ---------------------------------------------------------------------------
+
+
+class TestProtocolValidation:
+    """Validate :meth:`StrategyRegistry.validate_registry` behavior."""
+
+    def test_validate_returns_no_warnings_empty(self):
+        from cleveragents.application.services.strategy_registry import (
+            StrategyRegistry,
+        )
+
+        registry = StrategyRegistry()
+        warnings = registry.validate_registry()
+        assert warnings == []
+
+    def test_validate_enabled_but_not_registered_warns(self):
+        from cleveragents.application.services.strategy_registry import (
+            StrategyRegistry,
+        )
+
+        registry = StrategyRegistry()
+        registry.inject_stale_enabled_entry("ghost-strategy")
+        warnings = registry.validate_registry()
+        assert any("ghost-strategy" in w for w in warnings)
+
+    def test_validate_domain_strategy_no_resource_types_warns(self):
+        """Domain-model strategies must declare resource_types."""
+        from cleveragents.application.services.strategy_registry import (
+            StrategyRegistry,
+        )
+
+        registry = StrategyRegistry()
+        strat = FakeStrategyWithNoCaps()  # has domain caps with empty resource_types
+        registry.register(strat)
+        warnings = registry.validate_registry()
+        assert any("resource types" in w for w in warnings)
+
+    def test_validate_domain_strategy_with_resource_types_ok(self):
+        """Domain strategies that declare resource_types pass validation."""
+        from cleveragents.application.services.strategy_registry import (
+            StrategyRegistry,
+        )
+
+        registry = StrategyRegistry()
+        strat = FakeDomainStrategy("with-caps")  # has resource_types=("text",)
+        registry.register(strat)
+
+        warnings = registry.validate_registry()
+        assert not any("resource types" in w for w in warnings)
+
+    def test_validate_v1_pipeline_strategies_skip_resource_types(self):
+        """V1 pipeline strategies (with acms_service StrategyCapabilities) should NOT
+        get resource-types warnings because their capabilities object has no
+        such field.  This was the fix introduced by #10590."""
+        from cleveragents.application.services.strategy_registry import (
+            StrategyRegistry,
+        )
+
+        registry = StrategyRegistry()
+
+        # SimpleKeywordStrategy uses v1 pipeline StrategyCapabilities (acms_service)
+        # which lacks the domain-model resource_types field.
+        from cleveragents.application.services.context_strategies import (
+            SimpleKeywordStrategy,
+        )
+
+        registry.register(SimpleKeywordStrategy(), is_builtin=True)
+        warnings = registry.validate_registry()
+
+        # Should NOT have a "resource types" warning for v1 pipeline strategies
+        resource_warnings = [w for w in warnings if "resource types" in w]
+        assert len(resource_warnings) == 0
+
+
+# ---------------------------------------------------------------------------
+# Tests: Plugin loading from module:ClassName strings
+# ---------------------------------------------------------------------------
+
+
+class TestPluginLoading:
+    """Test :meth:`StrategyRegistry.register_from_module`."""
+
+    def test_register_from_module_success(self):
+        from cleveragents.application.services.context_strategies import (
+            StrategyConfig,
+            StrategyRegistry,
+        )
+
+        registry = StrategyRegistry()
+        registry.register_from_module(
+            name="simple-keyword",
+            module_path="cleveragents.application.services.context_strategies:SimpleKeywordStrategy",
+            config=StrategyConfig(enabled=True),
+        )
+        assert "simple-keyword" in registry
+        strategy = registry.get("simple-keyword")
+        assert strategy.name == "simple-keyword"
+
+    def test_register_from_module_invalid_format_raises(self):
+        from cleveragents.application.services.strategy_registry import (
+            StrategyRegistrationError,
+            StrategyRegistry,
+        )
+
+        registry = StrategyRegistry()
+        with pytest.raises(StrategyRegistrationError, match="module:ClassName"):
+            registry.register_from_module(name="bad", module_path="invalid_format")  # type: ignore[arg-type]
+
+    def test_register_from_module_unauthorized_prefix_raises(self):
+        from cleveragents.application.services.strategy_registry import (
+            StrategyRegistrationError,
+            StrategyRegistry,
+        )
+
+        registry = StrategyRegistry()
+        with pytest.raises(StrategyRegistrationError):
+            registry.register_from_module(
+                name="evil",
+                module_path="os.path:join",  # type: ignore[arg-type]
+            )
+
+    def test_register_from_module_nonexistent_class_raises(self):
+        from cleveragents.application.services.strategy_registry import (
+            StrategyRegistrationError,
+            StrategyRegistry,
+        )
+
+        registry = StrategyRegistry()
+        with pytest.raises(StrategyRegistrationError, match="not found"):
+            registry.register_from_module(
+                name="bogus",
+                module_path="cleveragents.application.services.context_strategies:"
+                "NoSuchClassDoesNotExist",  # type: ignore[arg-type]
+            )
+
+
+# ---------------------------------------------------------------------------
+# Tests: TOML configuration loading
+# ---------------------------------------------------------------------------
+
+
+class TestTOMLConfigurationLoading:
+    """Test :func:`load_strategies_from_config`.
+
+    This exercises the entire pipeline: register built-ins, optionally load
+    custom plugins from module:ClassName strings, and set the enabled list.
+    """
+
+    def test_load_empty_config_registers_defaults(self):
+        from cleveragents.application.services.context_strategies import (
+            load_strategies_from_config,
+        )
+
+        registry = load_strategies_from_config({})  # {}  = minimal config
+        names = registry.list_all()
+        assert "simple-keyword" in names
+        assert "semantic-embedding" in names
+        assert "breadth-depth-navigator" in names
+
+        # Only defaults should be enabled
+        enabled = registry.list_enabled()
+        assert set(enabled) == {
+            "simple-keyword",
+            "semantic-embedding",
+            "breadth-depth-navigator",
+        }
+
+    def test_load_config_with_custom_plugin(self):
+        from cleveragents.application.services.context_strategies import (
+            load_strategies_from_config,
+        )
+
+        registry = load_strategies_from_config(
+            {
+                "custom": {
+                    "my-strategy": "cleveragents.application.services.context_strategies:"
+                    "SemanticEmbeddingStrategy",
+                },
+            }
+        )
+        assert "my-strategy" in registry
+        assert "simple-keyword" in registry
+
+    def test_load_config_with_custom_and_custom_enabled(self):
+        from cleveragents.application.services.context_strategies import (
+            load_strategies_from_config,
+        )
+
+        registry = load_strategies_from_config(
+            {
+                "enabled": ["simple-keyword", "my-strategy"],
+                "custom": {
+                    "my-strategy": "cleveragents.application.services.context_strategies:"
+                    "SemanticEmbeddingStrategy",  # type: ignore[dict-item]
+                },
+            }
+        )
+        enabled = registry.list_enabled()
+        assert set(enabled) == {"simple-keyword", "my-strategy"}
+
+    def test_load_config_all_builtins_registered_as_builtin(self):
+        from cleveragents.application.services.context_strategies import (
+            load_strategies_from_config,
+        )
+
+        registry = load_strategies_from_config({})
+        for name in ["simple-keyword", "semantic-embedding", "breadth-depth-navigator"]:
+            entry = registry.get_entry(name)
+            assert entry.is_builtin is True
+
+
+# ---------------------------------------------------------------------------
+# Tests: Thread safety of registry operations
+# ---------------------------------------------------------------------------
+
+
+class TestThreadSafety:
+    """Verify that concurrent register/get/set_enabled calls don't corrupt state."""
+
+    def test_concurrent_register(self):
+        from cleveragents.application.services.strategy_registry import (
+            StrategyRegistry,
+        )
+
+        registry = StrategyRegistry()
+        errors: list[Exception] = []  # type: ignore[var-annotated]
+
+        def register_worker(worker_id: int) -> None:
+            try:
+                for _ in range(50):
+                    name = f"concurrent-strat-{worker_id}"
+                    registry.register(FakeStrategy(name))
+            except Exception as exc:  # noqa: BLE001
+                errors.append(exc)
+
+        threads = [threading.Thread(target=register_worker, args=(i,)) for i in range(10)]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join()
+
+        assert not errors, f"Errors during concurrent register: {errors}"
+        # Each worker registers once (deduplicated within its own run)
+        total = len(registry)
+        assert total > 0
+
+    def test_concurrent_get_and_register(self):
+        from cleveragents.application.services.strategy_registry import (
+            StrategyRegistry,
+        )
+
+        registry = StrategyRegistry()
+        registry.register(FakeStrategy("shared"))  # Pre-register one strategy
+        errors: list[Exception] = []  # type: ignore[var-annotated]
+        reads: list[str] = []  # type: ignore[var-annotated]
+
+        def reader(count: int) -> None:
+            for _ in range(20):
+                try:
+                    s = registry.get("shared")
+                    assert s is not None
+                    reads.append(s.name)
+                except Exception as exc:  # noqa: BLE001
+                    errors.append(exc)
+
+        def writer(worker_id: int) -> None:
+            for _ in range(20):
+                name = f"writer-{worker_id}"
+                registry.register(FakeStrategy(name))
+
+        threads: list[threading.Thread] = []
+        threads.append(threading.Thread(target=reader, args=(1,)))
+        for i in range(5):
+            threads.append(threading.Thread(target=writer, args=(i,)))
+
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join()
+
+        assert not errors
+        assert len(reads) == 20
+
+
+# ---------------------------------------------------------------------------
+# Tests: Fallback degradation chain verification
+# ---------------------------------------------------------------------------
+
+
+class TestFallbackDegradation:
+    """Verify that SimpleKeywordStrategy is always available as a fallback."""
+
+    def test_simple_keyword_always_produces_results(self):
+        from cleveragents.application.services.context_strategies import (
+            SimpleKeywordStrategy,
+        )
+        from cleveragents.domain.models.core.context_fragment import ContextBudget
+
+        strat = SimpleKeywordStrategy()
+        budget = ContextBudget(max_tokens=1000)
+
+        # Even with empty fragments, should return an empty list (no crash)
+        result = strat.assemble([], budget)
+        assert result == []
+
+        # Create a fake fragment-like object
+        class FakeFrag:  # pylint: disable=C0115
+            relevance_score: float = 0.5
+            token_count: int = 10
+            uko_node: str = "project://test/node"
+            content: str = "hello world test fragment"
+
+        fragments = [FakeFrag()]  # type: ignore[list-item]
+        result = strat.assemble(fragments, budget)
+        assert len(result) == 1
+
+    def test_simple_keyword_confidence_is_universal(self):
+        from cleveragents.application.services.context_strategies import (
+            SimpleKeywordStrategy,
+        )
+
+        strat = SimpleKeywordStrategy()
+        # Without query in request
+        assert strat.can_handle({}) == 0.3
+        # Request is always handled with at least 0.3 confidence
+        assert strat.can_handle({"query": ""}) == 0.3
+        assert strat.can_handle({"anything": "irrelevant"}) == 0.3
+
+
+# ---------------------------------------------------------------------------
+# Tests: Config-driven strategy enable/disable
+# ---------------------------------------------------------------------------
+
+
+class TestEnableDisable:
+    """Test :meth:`StrategyRegistry.set_enabled` and config updates."""
+
+    def test_set_enabled_updates_list(self):
+        from cleveragents.application.services.strategy_registry import (
+            StrategyConfig,
+            StrategyRegistry,
+        )
+
+        registry = StrategyRegistry()
+        for name in ["a", "b", "c"]:
+            registry.register(FakeStrategy(name), config=StrategyConfig(enabled=True))
+
+        # Only enable 'a' and 'c'
+        registry.set_enabled(["a", "c"])
+        enabled = registry.list_enabled()
+        assert enabled == ["a", "c"]
+
+    def test_set_unknown_strategy_raises(self):
+        from cleveragents.application.services.strategy_registry import (
+            StrategyNotFoundError,
+            StrategyRegistry,
+        )
+
+        registry = StrategyRegistry()
+        registry.register(FakeStrategy("exists"))
+        with pytest.raises(StrategyNotFoundError, match="unknown"):
+            registry.set_enabled(["nonexistent"])
+
+    def test_set_enabled_updates_config_flags(self):
+        from cleveragents.application.services.strategy_registry import (
+            StrategyConfig,
+            StrategyRegistry,
+        )
+
+        registry = StrategyRegistry()
+        for name in ["x", "y"]:
+            registry.register(FakeStrategy(name))
+
+        # Initially enabled by default (enable_by_default=True)
+        assert registry.get_config("x").enabled is True
+
+        registry.set_enabled(["x"])  # disable y only
+
+        # Check config entries: x stays enabled, y should be disabled
+        assert registry.get_config("x").enabled is True
+        assert registry.get_config("y").enabled is False
+
+    def test_update_config_individual_fields(self):
+        from cleveragents.application.services.strategy_registry import (
+            StrategyConfig,
+            StrategyRegistry,
+        )
+
+        registry = StrategyRegistry()
+        registry.register(FakeStrategy("updatable"))
+
+        registry.update_config(
+            "updatable",
+            enabled=False,
+            timeout_seconds=120,
+            max_fragments=50,
+        )
+
+        cfg = registry.get_config("updatable")
+        assert cfg.enabled is False
+        assert cfg.timeout_seconds == 120
+        assert cfg.max_fragments == 50
+
+    def test_update_config_enable_re_adds_to_enabled_order(self):
+        from cleveragents.application.services.strategy_registry import (
+            StrategyRegistry,
+        )
+
+        registry = StrategyRegistry()
+        registry.register(FakeStrategy("re-enable"))
+        registry.update_config("re-enable", enabled=True)
+
+        assert "re-enable" in registry._enabled_order  # pylint: disable=protected-access
+
+    def test_update_config_disable_removes_from_enabled_order(self):
+        from cleveragents.application.services.strategy_registry import (
+            StrategyRegistry,
+        )
+
+        registry = StrategyRegistry()
+        registry.register(FakeStrategy("disable-me"))
+        assert "disable-me" in registry._enabled_order  # pylint: disable=protected-access
+
+        registry.update_config("disable-me", enabled=False)
+        assert "disable-me" not in registry._enabled_order  # pylint: disable=protected-access
+
+
+# ---------------------------------------------------------------------------
+# Tests: Integration with acms_service auto-loading
+# ---------------------------------------------------------------------------
+
+
+class TestACMSPipelineAutoLoading:
+    """Verify that ACMSPipeline.__init__ loads strategies from Settings."""
+
+    def test_pipeline_auto_loads_strategies_from_settings(self):
+        from unittest.mock import MagicMock, PropertyMock
+
+        # Create a mock Settings object with context.strategies config.
+        settings = MagicMock()
+        settings.context = {  # type: ignore[attr-defined]
+            "strategies": {
+                "enabled": ["simple-keyword", "semantic-embedding"],
+            }
+        }
+
+        from cleveragents.application.services.acms_service import ACMSPipeline
+
+        pipeline = ACMSPipeline(settings=settings)
+        strategies_in_pipeline = list(pipeline._strategies.keys())  # pylint: disable=protected-access
+
+        # Built-in relevance strategy is always registered
+        assert "relevance" in strategies_in_pipeline
+        # Simple-keyword should be auto-loaded
+        assert "simple-keyword" in strategies_in_pipeline
+
+    def test_pipeline_works_without_settings(self):
+        from cleveragents.application.services.acms_service import ACMSPipeline
+
+        pipeline = ACMSPipeline()
+        assert "relevance" in pipeline._strategies  # pylint: disable=protected-access