sudacode/dotfiles
1
0 Fork 0
mirror of https://github.com/ksyasuda/dotfiles.git synced 2026-10-20 23:01:23 -07:00
Code Issues Packages Projects Releases Wiki Activity

update

Browse Source
This commit is contained in:
sudacode committed 2026-10-12 21:34:30 -07:00
1 parent 6cec9c3f70
commit 8f0b3f10e7
9 files changed
+1 -1026
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.agents/skills/delegate/SKILL.md
-187
View File
@@ -1,187 +0,0 @@
---
name: delegate
description: Delegate self-contained tasks to Claude or Codex subagents run as fresh CLI processes. Use proactively, without being asked, whenever work can be handed off — parallel research or code searches, bounded implementation chunks, independent reviews, second implementations for comparison. Pick the provider (Claude, Codex, or both) and tier that best fit each task. Not for consultations the user explicitly routed to a specific peer (use peer-consult for that).
---
# Delegate
Hand a self-contained task to a fresh `claude` or `codex` CLI process and get
back a structured report. Each invocation is one subagent; run several in
parallel for independent tasks.
You call the CLIs directly. The command shapes below are the default recipe,
not a fixed harness — adjust flags when the task needs it (extra directories,
different tools, a longer timeout), and say what you changed in your report.
## When to delegate
- Research or code searches whose details you don't need in your own context.
- Bounded implementation chunks that can be specified up front and verified.
- Independent review passes over a diff, file, or subsystem.
- Getting two independent takes by sending the same task to both providers.
Do not delegate work that needs mid-task input from you or the user, or tasks
so small that writing the brief costs more than doing the work.
## Route the task
Classify the smallest adequate tier:
| Tier | Signals | Codex | Claude |
|---|---|---|---|
| `quick` | Narrow, bounded, one component | `gpt-5.6-terra`, medium | `claude-sonnet-4-6`, medium |
| `quick-context` | Still bounded; more context or nuance | `gpt-5.6-terra`, high | `claude-sonnet-5`, high |
| `standard` | Multi-file feature, ordinary design/debugging, meaningful trade-offs | `gpt-5.6-sol`, medium | `claude-opus-4-8`, medium |
| `deep` | Architecture, concurrency, security, broad ambiguity, high-impact choice | `gpt-5.6-sol`, high | `claude-opus-4-8`, high |
Pick the provider by judgment:
- Either provider handles general coding, research, and review well; when the
task has no special pull, prefer the provider you are NOT running as, so the
subagent brings an independent perspective.
- Use **both** (two invocations, same task) when you want independent
perspectives on a risky or ambiguous problem, or when splitting parallel
work across providers to avoid rate-limit contention.
- Honor an explicit user model preference only if it is in the allowed list:
Codex `gpt-5.6-terra`/`gpt-5.6-sol`; Claude
`claude-sonnet-4-6`/`claude-sonnet-5`/`claude-opus-4-8`. Never silently
substitute another model.
Pick the mode:
- `read` (default): research, searches, reviews, second opinions. Subagent
cannot modify files.
- `write`: implementation and refactors. Subagent may edit files and run
commands inside the repo; it will not commit or push unless the task says to.
## Write the brief
The subagent starts with zero context. Write a self-contained task containing:
- The goal and definition of done.
- Relevant paths, symbols, error messages, or raw diff/plan text.
- Constraints (style, APIs to use or avoid, files off-limits).
- For `write` mode: how to verify (build/test commands).
Pass raw evidence; do not tell the subagent the answer you expect.
Wrap the brief in this scaffold. Keep the constraint block matching the mode
and keep the report sections verbatim — the result-handling steps below assume
them.
```
You are a subagent completing a delegated task inside the current repository.
Work independently; the caller cannot answer questions mid-task. If the task is
ambiguous, choose the safest reasonable interpretation and note the choice in
your report.
<constraints for the chosen mode — see below>
- Do not delegate further or invoke Claude or Codex recursively.
End your reply with a report containing these sections:
- Summary: what you did or found, in a few sentences.
- Details: key evidence, decisions, or findings, with file:line references.
- Files changed: list each changed file, or "none".
- Verification: what you ran and the outcome, or "not verified" and why.
- Open questions: anything unresolved the caller must decide, or "none".
Task:
<self-contained task brief>
```
`read` constraints:
```
Hard constraints:
- You are in read-only mode. Do not modify, create, rename, or delete files.
- Do not commit, branch, push, or contact external systems that change state.
- Base claims on repository evidence. Cite file paths and line numbers.
```
`write` constraints:
```
Hard constraints:
- Work only inside the given repository directory.
- Do not commit, branch, push, or open pull requests unless the task says to.
- Do not delete or rename files the task does not cover; if something looks
wrong or unexpected, stop and report instead of guessing.
- Verify your work (build, tests, or a targeted check) when feasible.
```
## Run it
Always send the prompt on stdin via a quoted heredoc (`<<'TASK'`) so nothing in
the brief is expanded or re-parsed by the shell. Run the CLI from the target
repo with `cd <repo> && ...`.
Codex, read-only:
```bash
cd /path/to/repo && codex exec \
--ignore-user-config \
--model gpt-5.6-sol \
--config model_reasoning_effort="medium" \
--config approval_policy="never" \
--sandbox read-only \
--ephemeral \
--skip-git-repo-check \
--output-last-message /tmp/delegate-codex-1.md \
--cd /path/to/repo - <<'TASK'
<scaffold + brief>
TASK
cat /tmp/delegate-codex-1.md
```
Codex, write mode: swap `--sandbox read-only` for `--sandbox workspace-write`.
`--output-last-message` writes just the final report; Codex's stdout also
carries its reasoning stream, which is useful when a run fails or you want to
see what it actually did. Add `--add-dir <dir>` for extra writable roots.
Claude, read-only:
```bash
cd /path/to/repo && claude --print \
--model claude-opus-4-8 \
--effort medium \
--disable-slash-commands \
--no-session-persistence \
--permission-mode plan \
--tools Read,Glob,Grep <<'TASK'
<scaffold + brief>
TASK
```
Claude, write mode: replace the last two lines with
```bash
--permission-mode acceptEdits \
--tools Read,Glob,Grep,Edit,Write,Bash \
--allowedTools Edit,Write,Bash
```
Claude prints the final message on stdout. Widen `--tools` when a task
genuinely needs more (for example `WebSearch` for external research, or
`Bash` in read mode for a `git diff` the subagent must see) — that is a
deliberate choice, so note it when you report back.
Give long runs a timeout that fits the task (the Bash tool's `timeout` is in
milliseconds; 1800000 is a reasonable ceiling for a `deep` tier run).
Parallel delegation: launch each invocation as a separate background shell
command, write each Codex report to its own `--output-last-message` path, then
collect the outputs. Never point two `write`-mode subagents at overlapping
files; split by file/directory or run them sequentially.
## Handle the result
- Treat the report as a subagent's claim, not ground truth: spot-check
load-bearing findings, and for `write` mode review the diff (`git diff`) and
re-run verification before building on it.
- Relay the outcome to the user in your own words; credit which provider/model
produced it when it matters.
- On failure, report the provider, model, and exact error. A nonzero exit or
empty output is a failure even if the CLI printed something — check both.
Retry once with the same route if transient; escalate tier or switch provider
only deliberately, and say you did.
.agents/skills/peer-consult/SKILL.md
-95
View File
@@ -1,95 +0,0 @@
---
name: peer-consult
description: Request an independent, read-only repository review or second opinion from the opposite model provider, using Claude from Codex or Codex from Claude. Use only when the user explicitly asks to consult, ask, review with, or get an opinion from Claude or Codex about an issue, feature, design, plan, diagnosis, or decision. Never invoke proactively. Never use for implementation or file changes.
---
# Peer Consult
Get one independent opinion from the opposite provider. Preserve the repository unchanged and return a standardized decision memo.
## Preconditions
- Require an explicit user request for cross-provider consultation. A normal request for review, planning, or advice is insufficient unless it names Claude, Codex, a peer model, or a second opinion.
- Run Claude only when the active agent is Codex. Run Codex only when the active agent is Claude.
- Do not use this skill from any other caller.
- Do not edit files, implement the recommendation, or perform follow-on changes unless the user separately requests them after seeing the memo.
## Route the request
Classify the smallest adequate tier:
| Tier | Signals | Codex peer | Claude peer |
|---|---|---|---|
| Quick | Narrow, bounded, one component | `gpt-5.6-terra`, medium | `claude-sonnet-4-6`, medium |
| Quick, context-heavy | Still bounded; more context or nuance | `gpt-5.6-terra`, high | `claude-sonnet-5`, high |
| Standard | Multi-file feature, ordinary design/debugging, meaningful trade-offs | `gpt-5.6-sol`, medium | `claude-opus-4-8`, medium |
| Deep | Architecture, concurrency, security, broad ambiguity, high-impact choice | `gpt-5.6-sol`, high | `claude-opus-4-8`, high |
Honor an explicit model override only when it is allowed for the target provider:
- Codex: `gpt-5.6-terra`, `gpt-5.6-sol`
- Claude: `claude-sonnet-4-6`, `claude-sonnet-5`, `claude-opus-4-8`
Reject GPT-5.6 Luna, pre-5.6 Codex models, Claude Haiku, every other Claude model, aliases, and cross-provider model names. Never silently fall back.
## Prepare independent context
Write a self-contained request containing:
- The question or decision to assess.
- Relevant constraints and success criteria.
- Relevant paths, symbols, errors, or raw plan text.
- Specific uncertainties the peer should resolve.
Pass raw evidence. Avoid telling the peer the expected answer or presenting the active agent's conclusion as fact. Let the peer inspect the current repository independently.
## Run the broker
Resolve `<skill-dir>` to the directory containing this `SKILL.md`. Send the request on stdin; do not interpolate it into a shell command.
From Codex:
```bash
python3 <skill-dir>/scripts/peer_consult.py \
--caller codex \
--complexity standard \
--repo "$PWD" <<'PEER_REQUEST'
<self-contained consultation request>
PEER_REQUEST
```
From Claude:
```bash
python3 <skill-dir>/scripts/peer_consult.py \
--caller claude \
--complexity standard \
--repo "$PWD" <<'PEER_REQUEST'
<self-contained consultation request>
PEER_REQUEST
```
Adjust only these routing flags when needed:
- `--complexity quick|standard|deep`
- `--context-heavy` for the context-heavy quick tier
- `--model <approved-model>` for an explicit allowed override
- `--timeout <seconds>` when the default 600 seconds is unsuitable
- `--dry-run` to inspect routing and command construction without calling a provider
The broker enforces pinned models, opposite-provider routing, nonpersistent execution, tool restrictions, structured output, and read-only repository access.
## Return the result
Return the broker's memo without changing its meaning. Keep its sections:
- Verdict
- Evidence
- Risks
- Alternatives
- Recommendation
- Confidence
Identify it as the peer provider's opinion. If useful, add a short, separately labeled synthesis explaining how the feedback affects the active discussion. Do not implement anything.
On failure, report the selected provider, model, and exact failure. Do not retry with another model unless the user explicitly requests it.
.agents/skills/peer-consult/agents/openai.yaml
-7
View File
@@ -1,7 +0,0 @@
interface:
display_name: "Peer Consult"
short_description: "Independent cross-provider second opinions"
default_prompt: "Use $peer-consult to get an independent read-only second opinion on this decision."
policy:
allow_implicit_invocation: true
.agents/skills/peer-consult/scripts/__pycache__/peer_consult.cpython-314.pyc
BIN
View File
Binary file not shown.
.agents/skills/peer-consult/scripts/memo.schema.json
-30
View File
@@ -1,30 +0,0 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"properties": {
"verdict": { "type": "string", "minLength": 1 },
"evidence": {
"type": "array",
"items": { "type": "string", "minLength": 1 }
},
"risks": {
"type": "array",
"items": { "type": "string", "minLength": 1 }
},
"alternatives": {
"type": "array",
"items": { "type": "string", "minLength": 1 }
},
"recommendation": { "type": "string", "minLength": 1 },
"confidence": { "type": "string", "enum": ["low", "medium", "high"] }
},
"required": [
"verdict",
"evidence",
"risks",
"alternatives",
"recommendation",
"confidence"
],
"additionalProperties": false
}
.agents/skills/peer-consult/scripts/peer_consult.py
-340
View File
@@ -1,340 +0,0 @@
#!/usr/bin/env python3
"""Run a read-only consultation with the opposite model provider."""
import argparse
import json
import os
import shutil
import subprocess
import sys
import tempfile
from dataclasses import dataclass
from pathlib import Path
ALLOWED_MODELS = {
"codex": frozenset({"gpt-5.6-terra", "gpt-5.6-sol"}),
"claude": frozenset(
{"claude-sonnet-4-6", "claude-sonnet-5", "claude-opus-4-8"}
),
}
def detect_caller(env: dict[str, str]) -> str:
is_codex = bool(env.get("CODEX_THREAD_ID"))
is_claude = env.get("CLAUDECODE") == "1" or bool(
env.get("CLAUDE_CODE_ENTRYPOINT")
)
if is_codex and is_claude:
raise ValueError("Ambiguous caller environment")
if is_codex:
return "codex"
if is_claude:
return "claude"
raise ValueError("Cannot detect caller; pass --caller claude or --caller codex")
@dataclass(frozen=True)
class Route:
provider: str
model: str
effort: str
def select_route(caller: str, complexity: str, context_heavy: bool = False) -> Route:
if complexity not in {"quick", "standard", "deep"}:
raise ValueError(f"Unknown complexity: {complexity}")
effort = (
"high"
if complexity == "deep" or (complexity == "quick" and context_heavy)
else "medium"
)
if caller == "claude":
model = "gpt-5.6-terra" if complexity == "quick" else "gpt-5.6-sol"
return Route("codex", model, effort)
if caller == "codex":
if complexity == "quick":
model = "claude-sonnet-5" if context_heavy else "claude-sonnet-4-6"
else:
model = "claude-opus-4-8"
return Route("claude", model, effort)
raise ValueError(f"Unknown caller: {caller}")
def apply_model_override(route: Route, model: str) -> Route:
if model not in ALLOWED_MODELS[route.provider]:
raise ValueError(f"Model {model!r} is not allowed for {route.provider}")
return Route(route.provider, model, route.effort)
def build_command(
route: Route,
repo: Path,
schema_path: Path,
output_path: Path | None = None,
) -> list[str]:
if route.provider == "codex":
if output_path is None:
raise ValueError("Codex requires an output path")
return [
"codex",
"exec",
"--ignore-user-config",
"--model",
route.model,
"--config",
f'model_reasoning_effort="{route.effort}"',
"--config",
'approval_policy="never"',
"--sandbox",
"read-only",
"--ephemeral",
"--skip-git-repo-check",
"--output-schema",
str(schema_path),
"--output-last-message",
str(output_path),
"--json",
"--cd",
str(repo),
"-",
]
schema = json.dumps(json.loads(schema_path.read_text()), separators=(",", ":"))
return [
"claude",
"--print",
"--safe-mode",
"--model",
route.model,
"--effort",
route.effort,
"--permission-mode",
"plan",
"--tools",
"Read,Glob,Grep",
"--disable-slash-commands",
"--no-session-persistence",
"--prompt-suggestions",
"false",
"--output-format",
"json",
"--json-schema",
schema,
]
def build_peer_prompt(question: str, route: Route) -> str:
return f"""You are an independent {route.provider} reviewer. Inspect the current repository in read-only mode and give a second opinion on the request below.
Hard constraints:
- Do not modify, create, rename, or delete files.
- Do not implement code or run commands that can change state.
- Do not commit, branch, push, open pull requests, or contact external systems.
- Do not delegate to subagents or invoke Claude, Codex, or peer-consult recursively.
- Base claims on repository evidence. Cite file paths and line numbers when available.
- Call out uncertainty and missing evidence. Do not pretend the caller's proposal is correct.
- Return only the structured decision memo required by the response schema.
Request:
{question.strip()}
"""
MEMO_FIELDS = {
"verdict",
"evidence",
"risks",
"alternatives",
"recommendation",
"confidence",
}
def validate_memo(value: object) -> dict[str, object]:
if not isinstance(value, dict) or set(value) != MEMO_FIELDS:
raise ValueError("Invalid decision memo fields")
if not isinstance(value["verdict"], str) or not value["verdict"].strip():
raise ValueError("Invalid decision memo verdict")
if not isinstance(value["recommendation"], str) or not value["recommendation"].strip():
raise ValueError("Invalid decision memo recommendation")
for field in ("evidence", "risks", "alternatives"):
items = value[field]
if not isinstance(items, list) or any(
not isinstance(item, str) or not item.strip() for item in items
):
raise ValueError(f"Invalid decision memo {field}")
if value["confidence"] not in {"low", "medium", "high"}:
raise ValueError("Invalid decision memo confidence")
return value
def parse_claude_output(output: str) -> object:
try:
envelope = json.loads(output)
except json.JSONDecodeError as error:
raise ValueError("Claude returned malformed JSON") from error
if isinstance(envelope, dict) and "structured_output" in envelope:
return envelope["structured_output"]
if isinstance(envelope, dict) and isinstance(envelope.get("result"), str):
try:
return json.loads(envelope["result"])
except json.JSONDecodeError as error:
raise ValueError("Claude result did not contain structured JSON") from error
return envelope
def _bullets(items: object) -> str:
values = items if isinstance(items, list) else []
return "\n".join(f"- {item}" for item in values) or "- None identified."
def render_memo(memo: dict[str, object], route: Route) -> str:
provider = route.provider.capitalize()
return f"""## Peer consultation
{provider} · {route.model} · {route.effort}
### Verdict
{memo['verdict']}
### Evidence
{_bullets(memo['evidence'])}
### Risks
{_bullets(memo['risks'])}
### Alternatives
{_bullets(memo['alternatives'])}
### Recommendation
{memo['recommendation']}
### Confidence
{memo['confidence']}
"""
def run_consultation(
route: Route,
repo: Path,
prompt: str,
schema_path: Path,
*,
timeout: int = 600,
env: dict[str, str] | None = None,
) -> dict[str, object]:
process_env = dict(os.environ if env is None else env)
if shutil.which(route.provider, path=process_env.get("PATH")) is None:
raise RuntimeError(f"Required CLI is not installed: {route.provider}")
with tempfile.TemporaryDirectory(prefix="peer-consult-") as temp_dir:
output_path = (
Path(temp_dir) / "memo.json" if route.provider == "codex" else None
)
command = build_command(route, repo, schema_path, output_path)
try:
result = subprocess.run(
command,
cwd=repo,
env=process_env,
input=prompt,
text=True,
capture_output=True,
timeout=timeout,
check=False,
)
except subprocess.TimeoutExpired as error:
raise RuntimeError(f"{route.provider} consultation timed out") from error
if result.returncode != 0:
detail = (result.stderr or result.stdout).strip() or "no error output"
raise RuntimeError(
f"{route.provider} consultation failed ({result.returncode}): {detail}"
)
if route.provider == "codex":
if output_path is None or not output_path.is_file():
raise RuntimeError("Codex did not produce a decision memo")
try:
value = json.loads(output_path.read_text())
except json.JSONDecodeError as error:
raise RuntimeError("Codex returned malformed JSON") from error
else:
value = parse_claude_output(result.stdout)
return validate_memo(value)
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="Get a read-only second opinion from the opposite model provider."
)
parser.add_argument("--caller", choices=("claude", "codex"))
parser.add_argument(
"--complexity",
choices=("quick", "standard", "deep"),
default="standard",
)
parser.add_argument("--context-heavy", action="store_true")
parser.add_argument("--model", help="Approved target-model override")
parser.add_argument("--repo", type=Path, default=Path.cwd())
parser.add_argument("--timeout", type=int, default=600)
parser.add_argument("--dry-run", action="store_true")
return parser.parse_args(argv)
def main(argv: list[str] | None = None) -> int:
args = parse_args(argv)
try:
question = sys.stdin.read().strip()
if not question:
raise ValueError("Consultation request must be provided on stdin")
repo = args.repo.expanduser().resolve()
if not repo.is_dir():
raise ValueError(f"Repository directory does not exist: {repo}")
caller = args.caller or detect_caller(dict(os.environ))
route = select_route(caller, args.complexity, args.context_heavy)
if args.model:
route = apply_model_override(route, args.model)
prompt = build_peer_prompt(question, route)
schema_path = Path(__file__).with_name("memo.schema.json")
output_path = (
Path(tempfile.gettempdir()) / "peer-consult-dry-run-output.json"
if route.provider == "codex"
else None
)
command = build_command(route, repo, schema_path, output_path)
if args.dry_run:
print(
json.dumps(
{
"provider": route.provider,
"model": route.model,
"effort": route.effort,
"command": command,
"prompt": prompt,
},
indent=2,
)
)
return 0
memo = run_consultation(
route,
repo,
prompt,
schema_path,
timeout=args.timeout,
)
print(render_memo(memo, route), end="")
return 0
except (OSError, RuntimeError, ValueError) as error:
print(f"peer-consult: error: {error}", file=sys.stderr)
return 2
if __name__ == "__main__":
raise SystemExit(main())
.agents/skills/peer-consult/tests/__pycache__/test_peer_consult.cpython-314.pyc
BIN
View File
Binary file not shown.
.agents/skills/peer-consult/tests/test_peer_consult.py
-366
View File
@@ -1,366 +0,0 @@
import importlib.util
import json
import os
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
SCRIPT = Path(__file__).parents[1] / "scripts" / "peer_consult.py"
def load_broker():
if not SCRIPT.exists():
raise AssertionError("peer_consult.py is missing")
spec = importlib.util.spec_from_file_location("peer_consult", SCRIPT)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
class RoutingTests(unittest.TestCase):
def test_routes_approved_models_and_effort(self):
broker = load_broker()
cases = {
("claude", "quick", False): ("codex", "gpt-5.6-terra", "medium"),
("claude", "quick", True): ("codex", "gpt-5.6-terra", "high"),
("claude", "standard", False): ("codex", "gpt-5.6-sol", "medium"),
("claude", "deep", False): ("codex", "gpt-5.6-sol", "high"),
("codex", "quick", False): ("claude", "claude-sonnet-4-6", "medium"),
("codex", "quick", True): ("claude", "claude-sonnet-5", "high"),
("codex", "standard", False): ("claude", "claude-opus-4-8", "medium"),
("codex", "deep", False): ("claude", "claude-opus-4-8", "high"),
}
for inputs, expected in cases.items():
with self.subTest(inputs=inputs):
route = broker.select_route(*inputs)
self.assertEqual(
(route.provider, route.model, route.effort),
expected,
)
def test_rejects_unknown_complexity(self):
broker = load_broker()
with self.assertRaisesRegex(ValueError, "Unknown complexity"):
broker.select_route("codex", "extreme")
def test_accepts_only_target_provider_model_overrides(self):
broker = load_broker()
self.assertTrue(
hasattr(broker, "apply_model_override"),
"model override validation is missing",
)
codex_route = broker.apply_model_override(
broker.select_route("claude", "quick"),
"gpt-5.6-sol",
)
claude_route = broker.apply_model_override(
broker.select_route("codex", "quick"),
"claude-opus-4-8",
)
self.assertEqual(codex_route.model, "gpt-5.6-sol")
self.assertEqual(claude_route.model, "claude-opus-4-8")
def test_rejects_disallowed_or_wrong_provider_model_overrides(self):
broker = load_broker()
self.assertTrue(
hasattr(broker, "apply_model_override"),
"model override validation is missing",
)
route = broker.select_route("claude", "quick")
for model in (
"gpt-5.6-luna",
"gpt-5.5",
"claude-sonnet-5",
"claude-haiku-4-5",
):
with self.subTest(model=model):
with self.assertRaisesRegex(ValueError, "not allowed"):
broker.apply_model_override(route, model)
class CallerDetectionTests(unittest.TestCase):
def test_detects_codex_and_claude_environments(self):
broker = load_broker()
self.assertTrue(hasattr(broker, "detect_caller"), "caller detection is missing")
self.assertEqual(broker.detect_caller({"CODEX_THREAD_ID": "abc"}), "codex")
self.assertEqual(broker.detect_caller({"CLAUDECODE": "1"}), "claude")
self.assertEqual(
broker.detect_caller({"CLAUDE_CODE_ENTRYPOINT": "cli"}),
"claude",
)
def test_rejects_missing_or_ambiguous_caller(self):
broker = load_broker()
self.assertTrue(hasattr(broker, "detect_caller"), "caller detection is missing")
with self.assertRaisesRegex(ValueError, "Cannot detect"):
broker.detect_caller({})
with self.assertRaisesRegex(ValueError, "Ambiguous"):
broker.detect_caller({"CODEX_THREAD_ID": "abc", "CLAUDECODE": "1"})
class CommandTests(unittest.TestCase):
def setUp(self):
self.broker = load_broker()
self.repo = Path("/tmp/example-repo")
self.schema = Path(__file__).parents[1] / "scripts" / "memo.schema.json"
def test_builds_read_only_ephemeral_codex_command(self):
self.assertTrue(hasattr(self.broker, "build_command"), "command builder is missing")
route = self.broker.select_route("claude", "standard")
command = self.broker.build_command(
route,
self.repo,
self.schema,
Path("/tmp/peer-output.json"),
)
self.assertEqual(command[:2], ["codex", "exec"])
self.assertIn("gpt-5.6-sol", command)
self.assertIn('model_reasoning_effort="medium"', command)
self.assertIn('approval_policy="never"', command)
self.assertIn("read-only", command)
self.assertIn("--ephemeral", command)
self.assertIn("--ignore-user-config", command)
self.assertIn("--json", command)
self.assertIn(str(self.schema), command)
self.assertIn("/tmp/peer-output.json", command)
self.assertEqual(command[-1], "-")
self.assertFalse(any("dangerously" in arg for arg in command))
def test_builds_safe_nonpersistent_claude_command(self):
self.assertTrue(hasattr(self.broker, "build_command"), "command builder is missing")
route = self.broker.select_route("codex", "deep")
command = self.broker.build_command(route, self.repo, self.schema)
self.assertEqual(command[0], "claude")
self.assertIn("--print", command)
self.assertIn("claude-opus-4-8", command)
self.assertIn("high", command)
self.assertIn("--safe-mode", command)
self.assertIn("--permission-mode", command)
self.assertIn("plan", command)
self.assertIn("--no-session-persistence", command)
self.assertIn("Read,Glob,Grep", command)
self.assertNotIn("Bash", command)
schema_arg = command[command.index("--json-schema") + 1]
self.assertEqual(json.loads(schema_arg)["additionalProperties"], False)
def test_peer_prompt_forbids_changes_and_requires_evidence(self):
self.assertTrue(hasattr(self.broker, "build_peer_prompt"), "peer prompt builder is missing")
route = self.broker.select_route("codex", "standard")
prompt = self.broker.build_peer_prompt("Should we split this service?", route)
for phrase in (
"read-only",
"Do not modify",
"Do not implement",
"Do not delegate",
"repository evidence",
"Should we split this service?",
):
with self.subTest(phrase=phrase):
self.assertIn(phrase, prompt)
class MemoTests(unittest.TestCase):
def setUp(self):
self.broker = load_broker()
self.memo = {
"verdict": "Split only after measuring coupling.",
"evidence": ["src/service.py:42 owns both workflows."],
"risks": ["A premature split adds coordination overhead."],
"alternatives": ["Extract an internal module first."],
"recommendation": "Instrument boundaries, then reassess.",
"confidence": "medium",
}
def test_validates_and_renders_standardized_memo(self):
self.assertTrue(hasattr(self.broker, "validate_memo"), "memo validation is missing")
self.assertTrue(hasattr(self.broker, "render_memo"), "memo rendering is missing")
route = self.broker.select_route("codex", "standard")
validated = self.broker.validate_memo(self.memo)
rendered = self.broker.render_memo(validated, route)
for heading in (
"## Peer consultation",
"### Verdict",
"### Evidence",
"### Risks",
"### Alternatives",
"### Recommendation",
"### Confidence",
):
with self.subTest(heading=heading):
self.assertIn(heading, rendered)
self.assertIn("Claude · claude-opus-4-8 · medium", rendered)
def test_rejects_malformed_memo(self):
self.assertTrue(hasattr(self.broker, "validate_memo"), "memo validation is missing")
malformed = dict(self.memo, confidence="certain", surprise="extra")
with self.assertRaisesRegex(ValueError, "memo"):
self.broker.validate_memo(malformed)
def test_extracts_claude_structured_output(self):
self.assertTrue(
hasattr(self.broker, "parse_claude_output"),
"Claude output parser is missing",
)
envelope = json.dumps({"type": "result", "structured_output": self.memo})
self.assertEqual(self.broker.parse_claude_output(envelope), self.memo)
class CliTests(unittest.TestCase):
def test_dry_run_prints_route_command_and_prompt_without_provider_call(self):
broker = load_broker()
self.assertTrue(hasattr(broker, "main"), "CLI entrypoint is missing")
with tempfile.TemporaryDirectory() as repo:
result = subprocess.run(
[
sys.executable,
str(SCRIPT),
"--caller",
"codex",
"--complexity",
"standard",
"--repo",
repo,
"--dry-run",
],
input="Should we split this service?",
text=True,
capture_output=True,
check=False,
)
self.assertEqual(result.returncode, 0, result.stderr)
payload = json.loads(result.stdout)
self.assertEqual(payload["provider"], "claude")
self.assertEqual(payload["model"], "claude-opus-4-8")
self.assertEqual(payload["effort"], "medium")
self.assertIn("--safe-mode", payload["command"])
self.assertIn("Should we split this service?", payload["prompt"])
def test_cli_rejects_luna_override_before_execution(self):
broker = load_broker()
self.assertTrue(hasattr(broker, "main"), "CLI entrypoint is missing")
result = subprocess.run(
[
sys.executable,
str(SCRIPT),
"--caller",
"claude",
"--complexity",
"quick",
"--model",
"gpt-5.6-luna",
"--dry-run",
],
input="Review this choice.",
text=True,
capture_output=True,
check=False,
)
self.assertNotEqual(result.returncode, 0)
self.assertIn("not allowed", result.stderr)
class ExecutionTests(unittest.TestCase):
def setUp(self):
self.broker = load_broker()
self.memo = {
"verdict": "Keep the boundary for now.",
"evidence": ["src/core.py:10 has one caller."],
"risks": ["The caller may grow."],
"alternatives": ["Revisit after instrumentation."],
"recommendation": "Measure first.",
"confidence": "high",
}
def _write_executable(self, path, source):
path.write_text(source)
path.chmod(0o755)
def test_runs_claude_and_extracts_structured_memo(self):
self.assertTrue(
hasattr(self.broker, "run_consultation"),
"consultation runner is missing",
)
with tempfile.TemporaryDirectory() as root:
root_path = Path(root)
bin_dir = root_path / "bin"
repo = root_path / "repo"
bin_dir.mkdir()
repo.mkdir()
envelope = json.dumps({"structured_output": self.memo})
self._write_executable(
bin_dir / "claude",
"#!/bin/sh\nread request\nprintf '%s' '" + envelope + "'\n",
)
env = dict(os.environ, PATH=f"{bin_dir}:{os.environ['PATH']}")
route = self.broker.select_route("codex", "standard")
memo = self.broker.run_consultation(
route,
repo,
"Review this design.",
SCRIPT.with_name("memo.schema.json"),
env=env,
)
self.assertEqual(memo, self.memo)
def test_runs_codex_and_reads_temporary_structured_memo(self):
self.assertTrue(
hasattr(self.broker, "run_consultation"),
"consultation runner is missing",
)
with tempfile.TemporaryDirectory() as root:
root_path = Path(root)
bin_dir = root_path / "bin"
repo = root_path / "repo"
bin_dir.mkdir()
repo.mkdir()
memo_json = json.dumps(self.memo)
self._write_executable(
bin_dir / "codex",
"#!/bin/sh\n"
"while [ \"$#\" -gt 0 ]; do\n"
" if [ \"$1\" = \"--output-last-message\" ]; then\n"
" shift\n"
f" printf '%s' '{memo_json}' > \"$1\"\n"
" exit 0\n"
" fi\n"
" shift\n"
"done\n"
"exit 9\n",
)
env = dict(os.environ, PATH=f"{bin_dir}:{os.environ['PATH']}")
route = self.broker.select_route("claude", "standard")
memo = self.broker.run_consultation(
route,
repo,
"Review this design.",
SCRIPT.with_name("memo.schema.json"),
env=env,
)
self.assertEqual(memo, self.memo)
if __name__ == "__main__":
unittest.main()