This repository was archived by the owner on Jun 20, 2026. It is now read-only.
Add Claude Code local orchestration bridge (bash & Python)#59
Merged
Conversation
- run-claude-task.sh: bash runner that accepts a task file or inline task string, passes it to `claude -p --no-interactive`, writes logs to claude-logs/, prints a summary table, and returns a meaningful exit code. Supports --queue mode for processing a whole directory of task files. Auto-detects gh and vercel CLIs. - run-claude-task.py: Python equivalent with structured JSONL logging, exponential-backoff retries, and --parallel flag for concurrent tasks. - task.txt: example task file demonstrating the --- separator syntax, comments, and GitHub/Vercel conditional steps. - CLAUDE-BRIDGE.md: full setup guide covering prerequisites, usage, environment variables, CI/GitHub Actions integration, and troubleshooting. https://claude.ai/code/session_01Eb18DymZ4mMwonPBcwpwXL
Reviewer's GuideAdds a local orchestration bridge for running Claude Code CLI tasks non-interactively, via a bash script and a more fully featured Python runner that support task files/queues, tool auto-detection, logging, retries, and parallel execution, plus documentation and an example task file. Sequence diagram for parallel task execution in the Python runnersequenceDiagram
actor Developer
participant PyRunner as run-claude-task.py
participant Tools as detect_tools
participant Exec as ThreadPoolExecutor
participant Worker1 as run_task(1)
participant Worker2 as run_task(2)
participant Claude as claude CLI
participant Logs as Log files (text + jsonl)
Developer->>PyRunner: Invoke with task.txt --parallel 2
PyRunner->>Tools: detect_tools()
Tools-->>PyRunner: {claude, gh?, vercel?, jq?}
PyRunner->>PyRunner: parse_task_file(task.txt) -> tasks[1..N]
PyRunner->>Exec: submit run_task for each task (max_workers=2)
par First batch
Exec->>Worker1: run_task(task1, index=1, label=file:1)
Exec->>Worker2: run_task(task2, index=2, label=file:2)
end
Worker1->>Claude: claude --model MODEL --no-interactive -p full_prompt(task1)
Worker2->>Claude: claude --model MODEL --no-interactive -p full_prompt(task2)
Claude-->>Worker1: exit_code, stdout, stderr
Claude-->>Worker2: exit_code, stdout, stderr
Worker1->>Worker1: handle retries / timeout
Worker2->>Worker2: handle retries / timeout
Worker1->>Logs: Append text log and JSONL record
Worker2->>Logs: Append text log and JSONL record
Worker1-->>Exec: TaskResult(1,...)
Worker2-->>Exec: TaskResult(2,...)
Exec-->>PyRunner: list[TaskResult]
PyRunner->>PyRunner: print_summary(results)
PyRunner-->>Developer: Exit with 0 if all ok, else non-zero
Class diagram for TaskResult dataclass in the Python runnerclassDiagram
class TaskResult {
+int index
+str label
+str status
+int exit_code
+float duration_s
+str output
+int retries
+str error
}
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 2 security issues, 3 other issues, and left some high level feedback:
Security issues:
- Detected subprocess function 'run' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'. (link)
- Detected subprocess function 'run' with user controlled data. A malicious actor could leverage this to perform command injection. You may consider using 'shlex.quote()'. (link)
General comments:
- In the bash runner,
set -ecombined withoutput=$(...) || exit_code=$?can cause the script to exit before the||handler runs whentimeoutreturns non-zero; consider restructuring (e.g. using a subshell that locally disablesset -e, or capturing$?inside the command substitution) so failures are reliably captured instead of aborting the whole script unexpectedly. - The bash script assumes GNU
timeoutis available, which is not present by default on some systems (e.g. macOS); you may want to either detect its presence and provide a clearer error or fall back to an alternative mechanism for enforcing per-task timeouts.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In the bash runner, `set -e` combined with `output=$(...) || exit_code=$?` can cause the script to exit before the `||` handler runs when `timeout` returns non-zero; consider restructuring (e.g. using a subshell that locally disables `set -e`, or capturing `$?` inside the command substitution) so failures are reliably captured instead of aborting the whole script unexpectedly.
- The bash script assumes GNU `timeout` is available, which is not present by default on some systems (e.g. macOS); you may want to either detect its presence and provide a clearer error or fall back to an alternative mechanism for enforcing per-task timeouts.
## Individual Comments
### Comment 1
<location path="run-claude-task.py" line_range="165" />
<code_context>
+ label: str,
+ tools_available: dict[str, str],
+ *,
+ retries: int = MAX_RETRY,
+) -> TaskResult:
+ """Execute one task via `claude -p` and return a TaskResult."""
</code_context>
<issue_to_address>
**issue (bug_risk):** Default `retries` value is bound at import time and ignores `--retries` overrides.
Since `retries` defaults to `MAX_RETRY`, its value is fixed at import time, before `main()` updates `MAX_RETRY` from CLI args. Calls that omit `retries` won’t honor `--retries`. To make the flag effective, pass a sentinel (e.g. `retries: Optional[int] = None`) and inside the function set `retries = MAX_RETRY if retries is None else retries`.
</issue_to_address>
### Comment 2
<location path="run-claude-task.py" line_range="73-76" />
<code_context>
+_jsonl_path = LOG_DIR / f"session-{SESSION_ID}.jsonl"
+
+
+def _jsonl(record: dict) -> None:
+ """Append a JSON line to the structured log."""
+ with _jsonl_path.open("a", encoding="utf-8") as fh:
+ fh.write(json.dumps(record, ensure_ascii=False) + "\n")
+
+
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Concurrent JSONL writes can interleave across threads and produce corrupted lines.
In parallel mode, multiple threads call `_jsonl`, each independently opening the same file and appending. Without synchronization, these writes can interleave at the OS level and produce mixed or truncated JSON objects. To keep JSONL machine-parseable, route all writes through a single shared file handle protected by a `threading.Lock`, or centralize logging on one thread (e.g., via a queue).
Suggested implementation:
```python
log = logging.getLogger("orchestrator")
# JSONL structured log file for machine consumption
_jsonl_path = LOG_DIR / f"session-{SESSION_ID}.jsonl"
_jsonl_lock = threading.Lock()
_jsonl_file = _jsonl_path.open("a", encoding="utf-8")
def _jsonl(record: dict) -> None:
"""Append a JSON line to the structured log in a thread-safe way."""
line = json.dumps(record, ensure_ascii=False) + "\n"
with _jsonl_lock:
_jsonl_file.write(line)
_jsonl_file.flush()
```
To fully implement this safely and avoid interleaved JSONL writes:
1. Ensure the following imports exist near the top of `run-claude-task.py`:
- `import threading`
- `import json`
2. Optionally, register a shutdown hook to close the shared handle cleanly, e.g.:
```python
import atexit
atexit.register(_jsonl_file.close)
```
3. Confirm that all structured logging in the file calls this `_jsonl(record: dict)` function and does not open/write to `_jsonl_path` directly.
</issue_to_address>
### Comment 3
<location path="CLAUDE-BRIDGE.md" line_range="159-168" />
<code_context>
+
+## Logs
+
+After each session two files are created inside `claude-logs/`:
+
+```
+claude-logs/
+ session-20260308-143012-12345.log # plain-text, human readable
+ session-20260308-143012-12345.jsonl # structured JSON lines (Python only)
+ summary-20260308-143012-12345.txt # one-line per task (bash only)
+```
+
</code_context>
<issue_to_address>
**issue (typo):** The text says two files are created, but three files are listed in the example.
The text conflicts with the example: it says two files are created but shows three (`session-*.log`, `session-*.jsonl`, `summary-*.txt`). Please update either the count or the list so they’re consistent.
```suggestion
## Logs
After each session, the following log files may be created inside `claude-logs/`:
```
claude-logs/
session-20260308-143012-12345.log # plain-text, human readable
session-20260308-143012-12345.jsonl # structured JSON lines (Python runner only)
summary-20260308-143012-12345.txt # one-line per task (bash runner only)
```
```
</issue_to_address>
### Comment 4
<location path="run-claude-task.py" line_range="205-211" />
<code_context>
result = subprocess.run(
cmd,
cwd=str(WORKDIR),
capture_output=True,
text=True,
timeout=TIMEOUT,
)
</code_context>
<issue_to_address>
**security (python.lang.security.audit.dangerous-subprocess-use-audit):** Detected subprocess function 'run' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'.
*Source: opengrep*
</issue_to_address>
### Comment 5
<location path="run-claude-task.py" line_range="206" />
<code_context>
cmd,
</code_context>
<issue_to_address>
**security (python.lang.security.audit.dangerous-subprocess-use-tainted-env-args):** Detected subprocess function 'run' with user controlled data. A malicious actor could leverage this to perform command injection. You may consider using 'shlex.quote()'.
*Source: opengrep*
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| label: str, | ||
| tools_available: dict[str, str], | ||
| *, | ||
| retries: int = MAX_RETRY, |
There was a problem hiding this comment.
issue (bug_risk): Default retries value is bound at import time and ignores --retries overrides.
Since retries defaults to MAX_RETRY, its value is fixed at import time, before main() updates MAX_RETRY from CLI args. Calls that omit retries won’t honor --retries. To make the flag effective, pass a sentinel (e.g. retries: Optional[int] = None) and inside the function set retries = MAX_RETRY if retries is None else retries.
Comment on lines
+73
to
+76
| def _jsonl(record: dict) -> None: | ||
| """Append a JSON line to the structured log.""" | ||
| with _jsonl_path.open("a", encoding="utf-8") as fh: | ||
| fh.write(json.dumps(record, ensure_ascii=False) + "\n") |
There was a problem hiding this comment.
suggestion (bug_risk): Concurrent JSONL writes can interleave across threads and produce corrupted lines.
In parallel mode, multiple threads call _jsonl, each independently opening the same file and appending. Without synchronization, these writes can interleave at the OS level and produce mixed or truncated JSON objects. To keep JSONL machine-parseable, route all writes through a single shared file handle protected by a threading.Lock, or centralize logging on one thread (e.g., via a queue).
Suggested implementation:
log = logging.getLogger("orchestrator")
# JSONL structured log file for machine consumption
_jsonl_path = LOG_DIR / f"session-{SESSION_ID}.jsonl"
_jsonl_lock = threading.Lock()
_jsonl_file = _jsonl_path.open("a", encoding="utf-8")
def _jsonl(record: dict) -> None:
"""Append a JSON line to the structured log in a thread-safe way."""
line = json.dumps(record, ensure_ascii=False) + "\n"
with _jsonl_lock:
_jsonl_file.write(line)
_jsonl_file.flush()To fully implement this safely and avoid interleaved JSONL writes:
- Ensure the following imports exist near the top of
run-claude-task.py:import threadingimport json
- Optionally, register a shutdown hook to close the shared handle cleanly, e.g.:
import atexit atexit.register(_jsonl_file.close)
- Confirm that all structured logging in the file calls this
_jsonl(record: dict)function and does not open/write to_jsonl_pathdirectly.
Comment on lines
+205
to
+211
| result = subprocess.run( | ||
| cmd, | ||
| cwd=str(WORKDIR), | ||
| capture_output=True, | ||
| text=True, | ||
| timeout=TIMEOUT, | ||
| ) |
There was a problem hiding this comment.
security (python.lang.security.audit.dangerous-subprocess-use-audit): Detected subprocess function 'run' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'.
Source: opengrep
|
|
||
| try: | ||
| result = subprocess.run( | ||
| cmd, |
There was a problem hiding this comment.
security (python.lang.security.audit.dangerous-subprocess-use-tainted-env-args): Detected subprocess function 'run' with user controlled data. A malicious actor could leverage this to perform command injection. You may consider using 'shlex.quote()'.
Source: opengrep
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
| label: str, | ||
| tools_available: dict[str, str], | ||
| *, | ||
| retries: int = MAX_RETRY, |
There was a problem hiding this comment.
🔴 Python default argument captures stale MAX_RETRY, making --retries CLI flag ineffective
The run_task function at run-claude-task.py:165 uses retries: int = MAX_RETRY as a default parameter. Python evaluates default parameter values once at function definition time (module load), not at call time. When main() updates the global MAX_RETRY = args.retries at run-claude-task.py:391, this does NOT update the already-captured default value. Since both run_tasks_sequential (run-claude-task.py:266) and run_tasks_parallel (run-claude-task.py:274) call run_task without passing the retries keyword argument, the --retries CLI flag is silently ignored and the retry count always uses the initial module-level value (from the environment variable or its default of 2).
Prompt for agents
In run-claude-task.py, the default parameter `retries: int = MAX_RETRY` at line 165 captures the value at function definition time, not at call time. To fix this, either:
1. Change line 165 to use a sentinel default and resolve inside the function body:
- Change `retries: int = MAX_RETRY` to `retries: int | None = None`
- Add at the top of run_task body: `if retries is None: retries = MAX_RETRY`
2. Or pass `retries=MAX_RETRY` explicitly from the callers:
- In run_tasks_sequential (line 266): `run_task(task, i, label, tools, retries=MAX_RETRY)`
- In run_tasks_parallel (line 274): `pool.submit(run_task, task, i + 1, label, tools, retries=MAX_RETRY)` — but note this won't work for keyword-only args with pool.submit, so use a lambda or partial.
Option 1 is simpler and more robust.
Was this helpful? React with 👍 or 👎 to provide feedback.
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: e11a55fe09
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| label: str, | ||
| tools_available: dict[str, str], | ||
| *, | ||
| retries: int = MAX_RETRY, |
There was a problem hiding this comment.
--retries is parsed and assigned to MAX_RETRY in main(), but run_task captures MAX_RETRY as a default argument at definition time, so CLI overrides are ignored whenever they differ from the startup env value. In practice, running run-claude-task.py ... --retries N still uses the original default retry count, which can make transient failures fail earlier than the user requested.
Useful? React with 👍 / 👎.
New entrypoint: claude-run (Python, zero external deps)
- Single command: ./claude-run task.txt
- Multi-task files with --- separators
- Inline task strings
- --dry-run: preview tasks without execution
- --retry N: exponential back-off retries (2s / 4s / 8s)
- --dangerously-skip-permissions: pass-through to Claude CLI
- --parallel N: concurrent task execution via ThreadPoolExecutor
- --fail-fast: abort on first failure
- --watch DIR: daemon that polls a folder for incoming .txt files,
moves completed files to tasks/done/, failures to tasks/failed/
- --queue DIR: one-shot batch over all *.txt in a folder
- --workdir / --log-dir / --model / --timeout / --poll flags
Logging:
- logs/ directory with timestamped per-session .log files
- Per-task individual .log files (stdout + stderr + metadata)
- JSONL machine-readable log per session with structured events:
session_start, task_result, session_end, watch_* events
Tool detection:
- Detects claude, gh, vercel, node, npm, jq, git, python3
- Injects tool availability context into every Claude prompt
- Clear warnings for missing optional tools (gh, vercel)
Example task files:
- tasks/cleanup.txt: GitHub issue triage, stale branch audit,
merged-PR branch cleanup report, open PR health matrix
- tasks/deploy.txt: Next.js build, test suite, Vercel deployment
history, preview deploy, PR comment with deploy URL
- tasks/audit.txt: unused dependency scan, TS strict-mode gap,
env var coverage, security pattern scan, large file report
Directory structure:
- tasks/done/ completed task files (watch mode)
- tasks/failed/ failed task files (watch mode)
- logs/ session + task logs
Docs: CLAUDE-BRIDGE.md fully rewritten with quick-start, all flag
reference, environment variables, JSONL schema, CI/GitHub Actions
example, watch-mode walk-through, dry-run, retry, and troubleshooting.
https://claude.ai/code/session_01Eb18DymZ4mMwonPBcwpwXL
Comment on lines
+285
to
+291
| proc = subprocess.run( | ||
| cmd, | ||
| cwd=str(WORKDIR), | ||
| capture_output=True, | ||
| text=True, | ||
| timeout=timeout, | ||
| ) |
There was a problem hiding this comment.
security (python.lang.security.audit.dangerous-subprocess-use-audit): Detected subprocess function 'run' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'.
Source: opengrep
Comment on lines
+301
to
+308
| except subprocess.TimeoutExpired: | ||
| last_exit = 124 | ||
| last_error = f"timed out after {timeout}s" | ||
| break # don't retry timeouts | ||
|
|
||
| except Exception as exc: # noqa: BLE001 | ||
| last_exit = -1 | ||
| last_error = str(exc) |
There was a problem hiding this comment.
🟡 Stale stdout/stderr recorded in task logs when timeout occurs on a retry attempt
In claude-run, when subprocess.TimeoutExpired is caught (line 301-304) or a generic Exception is caught (line 306-308), last_out and last_err are not reset. If the timeout/exception occurs on a retry attempt (attempt > 1), these variables retain the stdout/stderr from the previous failed attempt. The resulting TaskResult, per-task log file (line 355-357), and JSONL emission will all incorrectly attribute the prior attempt's output to the timed-out/errored attempt. Compare with run-claude-task.py:221 which correctly resets last_output = "" on timeout.
Suggested change
| except subprocess.TimeoutExpired: | |
| last_exit = 124 | |
| last_error = f"timed out after {timeout}s" | |
| break # don't retry timeouts | |
| except Exception as exc: # noqa: BLE001 | |
| last_exit = -1 | |
| last_error = str(exc) | |
| except subprocess.TimeoutExpired: | |
| last_exit = 124 | |
| last_out = "" | |
| last_err = "" | |
| last_error = f"timed out after {timeout}s" | |
| break # don't retry timeouts | |
| except Exception as exc: # noqa: BLE001 | |
| last_exit = -1 | |
| last_out = "" | |
| last_err = "" | |
| last_error = str(exc) |
Was this helpful? React with 👍 or 👎 to provide feedback.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to subscribe to this conversation on GitHub.
Already have an account?
Sign in.
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Introduces a local orchestration bridge for running Claude Code CLI tasks unattended from task files or command queues. Provides both a lightweight bash implementation and a feature-rich Python alternative with structured logging, automatic retries, and parallel execution support.
Key Changes
run-claude-task.sh— Bash runner script with zero Python dependency.txtfiles with---separators and#comment support--queuedirectory)gh,vercel,jq, etc.) and passes context to Clauderun-claude-task.py— Python runner with advanced features--retries)--parallel Nflag usingThreadPoolExecutor)*.jsonl) for machine parsingCLAUDE-BRIDGE.md— Comprehensive documentationjqexamplestask.txt— Example task file demonstrating:---separatorsgh,vercel)Implementation Details
claude -p --no-interactiveto prevent interactive promptsCLAUDE_LOG_DIR,CLAUDE_WORKDIR,CLAUDE_MODEL,CLAUDE_TIMEOUT,CLAUDE_RETRIESsubprocess.run()with timeout handling and structured error trackingtimeoutcommand and simple string accumulation for task parsinghttps://claude.ai/code/session_01Eb18DymZ4mMwonPBcwpwXL
Summary by Sourcery
Introduce a local orchestration bridge for running Claude Code CLI tasks non-interactively from task files, inline prompts, or queued directories, with both bash and Python entrypoints.
New Features:
Enhancements:
Documentation: