S11. Background Tasks — Slow Operations Go to the Background
S11. Background Tasks — Slow Operations Go to the Background
The Problem
Reading a file or running git status usually returns quickly, so synchronous execution causes little noticeable delay. Installing dependencies, running a full test suite, or building a project can take several minutes. Until the command returns, the Harness cannot process the next tool call in the current response or start the next model turn.
If later work does not depend on that command, there is no need to block it. For example, after starting a full test suite, the Agent could inspect documentation or organize other files while the tests run.
S11 addresses this by running slow Bash commands in the background, allowing the Agent Loop to continue and collect completed results on a later turn.
The Solution
This chapter sends slow operations to background threads. The current tool call first returns a placeholder tool_result, allowing the Agent Loop to continue. At the start of a later turn, completed results are collected and added to the conversation as notifications.
Sync vs Background:
| | Sync (s04) | Background (s11) | |---|---|---| | Slow operations | Current tool call blocks | Background thread executes | | Agent Loop | Waits for the command to return | Continues after the placeholder result | | Result | Returned after the command finishes | Returns bg_id first; collects the result on a later turn | | Decision criteria | — | bash run_in_background parameter |
How It Works
should_run_background: Explicit Request
The model requests background execution through the bash tool's run_in_background parameter. Only bash calls with the parameter explicitly set to true enter this path. Other calls still run synchronously.
def should_run_background(tool_name: str, tool_input: dict) -> bool:
return (
tool_name == "bash"
and tool_input.get("run_in_background") is True
)The Harness no longer guesses from keywords such as install, build, or test. The tool call chooses the execution mode explicitly.
BackgroundManager: Background Execution and Lifecycle
BackgroundManager owns task state and the completion queue. start() registers a task, starts a daemon thread, and returns bg_id immediately:
class BackgroundManager:
def __init__(self):
self.tasks = {}
self.results = {}
self._ready = []
self._lock = threading.Lock()
def start(self, block) -> str:
# Register task, then run _run() in a daemon thread.
...
def _run(self, task_id: str, command: str):
output, exit_code = _run_bash_process(command)
status = "completed" if exit_code == 0 else "failed"
with self._lock:
self.tasks[task_id]["status"] = status
self.results[task_id] = _format_bash_result(output, exit_code)
self._ready.append(task_id)A non-zero exit code or worker exception becomes failed. The shell starts in its own process group. When the command finishes, times out, or the Agent exits through the normal or SIGTERM path, the runtime stops that original group. This is lifecycle cleanup, not a sandbox: a process that creates another session can leave the group.
collect_background_results: Notification Collection
At the start of a later turn, collect() removes completed results from the queue and formats them as <task_notification> messages:
def collect_background_results() -> list[str]:
return BACKGROUND.collect()Notifications don't reuse the original tool_use_id. The original tool call was already answered with a placeholder tool_result; when the completed result is collected, it is added as an independent event in task_notification format. One tool_use still gets exactly one tool_result.
Loop Integration
Before each LLM call, the Agent Loop collects completed background results. execute_tool() still runs PreToolUse on the main thread before choosing synchronous or background execution:
while True:
inject_background_results(messages)
response = client.messages.create(...)
def execute_tool(block) -> str:
blocked = trigger_hooks("PreToolUse", block)
if blocked is not None:
return str(blocked)
if should_run_background(block.name, block.input):
task_id = start_background_task(block)
output = f"[Background task {task_id} started]"
else:
output = call_tool(block)
trigger_hooks("PostToolUse", block, output)
return outputSlow operations first return a placeholder tool_result with bg_id. A completed task does not wake the Agent by itself; inject_background_results() collects it the next time the Agent Loop runs.
Putting It Together
Turn 1:
LLM → bash "npm install" (run_in_background=true)
→ start_background_task → bg_0001
→ tool_result: "[Background task bg_0001 started]..."
→ LLM: "OK, I'll check later. Let me also read the config."
Turn 2:
LLM → read_file "package.json" (fast, sync)
→ tool_result: file content
Turn 3:
→ collect bg_0001 as <task_notification>
→ LLM sees: config file + install notification in one messageWhile npm install ran in the background, the Agent Loop continued with read_file.
What s11 Adds
| Component | s04 Kernel | s11 | |-----------|-------------|-------------| | Execution model | All synchronous | Slow ops to background thread + notification injection | | bash schema | command | command + run_in_background | | New functions | — | should_run_background, start_background_task, collect_background_results, inject_background_results | | New types | — | BackgroundManager | | Notification format | — | <task_notification> (doesn't reuse tool_use_id) | | Loop behavior | Tools execute synchronously | Explicit background execution, completed results collected on later turns | | Tools | 5 | 5 (one parameter added to the bash schema) |
Try It
cd learn-claude-code
python s11_background_tasks/code.pyTry these prompts:
Run pip list in the background and find all Python files in this directoryRun npm install (use run_in_background) and while waiting, read package.jsonRun a short sleep in the background, then list all Markdown files
What to observe: After explicitly setting run_in_background, is the command dispatched to the background? Is a bg_id returned? Are completed results collected in <task_notification> format on a later turn?
What's Next
Background tasks solved "slow operations don't block." But what if you want to do something on a schedule? Like "run tests every morning at 9am" or "check server status every 5 minutes."
s12 Cron Scheduler → Give the agent an alarm clock.
S11 — Complete teaching code
#!/usr/bin/env python3
"""
s11_background_tasks.py - Background Tasks
Main thread Background thread
+------------------------------+ +----------------------+
| bash(run_in_background=True) | ------> | run command |
| return bg_id | | queue result |
| continue agent loop | <------ +----------------------+
| next turn: collect |
+------------------------------+
"""
import atexit
import glob
import os
import signal
import subprocess
import threading
import time
from pathlib import Path
try:
import readline
readline.parse_and_bind("set bind-tty-special-chars off")
readline.parse_and_bind("set input-meta on")
readline.parse_and_bind("set output-meta on")
readline.parse_and_bind("set convert-meta off")
except ImportError:
pass
from anthropic import Anthropic
from dotenv import load_dotenv
load_dotenv(override=True)
if os.getenv("ANTHROPIC_BASE_URL"):
os.environ.pop("ANTHROPIC_AUTH_TOKEN", None)
WORKDIR = Path.cwd()
client = Anthropic(base_url=os.getenv("ANTHROPIC_BASE_URL"))
MODEL = os.environ["MODEL_ID"]
SYSTEM = (
f"You are a coding agent at {WORKDIR}. Use tools to solve tasks. "
"Set run_in_background to true only for independent Bash commands."
)
# -- From s04: tool implementations --
_shell_processes: set[subprocess.Popen] = set()
_shell_process_lock = threading.RLock()
def _stop_process_group(process: subprocess.Popen):
"""Stop processes that remain in the command's original process group."""
for sig in (signal.SIGTERM, signal.SIGKILL):
try:
os.killpg(process.pid, sig)
except (ProcessLookupError, OSError):
return
time.sleep(0.05)
def _stop_all_shell_processes():
with _shell_process_lock:
processes = list(_shell_processes)
for process in processes:
_stop_process_group(process)
def _handle_termination_signal(signum, _frame):
_stop_all_shell_processes()
raise SystemExit(128 + signum)
atexit.register(_stop_all_shell_processes)
signal.signal(signal.SIGTERM, _handle_termination_signal)
def _run_bash_process(command: str) -> tuple[str, int | None]:
process = None
try:
process = subprocess.Popen(
command,
shell=True,
cwd=WORKDIR,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
start_new_session=True,
)
with _shell_process_lock:
_shell_processes.add(process)
stdout, stderr = process.communicate(timeout=120)
output = (stdout + stderr).strip()
return (output[:50000] if output else "(no output)"), process.returncode
except subprocess.TimeoutExpired:
return "Error: Timeout (120s)", None
except OSError as error:
return f"Error: {type(error).__name__}: {error}", None
finally:
if process is not None:
_stop_process_group(process)
try:
process.wait(timeout=0.2)
except subprocess.TimeoutExpired:
pass
with _shell_process_lock:
_shell_processes.discard(process)
def _format_bash_result(output: str, exit_code: int | None) -> str:
if exit_code in (0, None):
return output
return f"Error: command exited with status {exit_code}\n{output}"
def run_bash(command: str, run_in_background: bool = False) -> str:
return _format_bash_result(*_run_bash_process(command))
def run_read(path: str, limit: int | None = None) -> str:
try:
file_path = (WORKDIR / path).resolve()
lines = file_path.read_text().splitlines()
if limit and limit < len(lines):
lines = lines[:limit] + [f"... ({len(lines) - limit} more lines)"]
return "\n".join(lines)
except Exception as error:
return f"Error: {error}"
def run_write(path: str, content: str) -> str:
try:
file_path = (WORKDIR / path).resolve()
file_path.parent.mkdir(parents=True, exist_ok=True)
file_path.write_text(content)
return f"Wrote {len(content)} bytes to {path}"
except Exception as error:
return f"Error: {error}"
def run_edit(path: str, old_text: str, new_text: str) -> str:
try:
file_path = (WORKDIR / path).resolve()
text = file_path.read_text()
if old_text not in text:
return f"Error: text not found in {path}"
file_path.write_text(text.replace(old_text, new_text, 1))
return f"Edited {path}"
except Exception as error:
return f"Error: {error}"
def run_glob(pattern: str) -> str:
try:
matches = [
match
for match in glob.glob(pattern, root_dir=WORKDIR)
if (WORKDIR / match).resolve().is_relative_to(WORKDIR)
]
return "\n".join(matches) if matches else "(no matches)"
except Exception as error:
return f"Error: {error}"
TOOLS = [
{"name": "bash", "description": "Run a shell command.",
"input_schema": {"type": "object",
"properties": {
"command": {"type": "string"},
"run_in_background": {"type": "boolean"}},
"required": ["command"]}},
{"name": "read_file", "description": "Read file contents.",
"input_schema": {"type": "object",
"properties": {"path": {"type": "string"},
"limit": {"type": "integer"}},
"required": ["path"]}},
{"name": "write_file", "description": "Write content to a file.",
"input_schema": {"type": "object",
"properties": {"path": {"type": "string"},
"content": {"type": "string"}},
"required": ["path", "content"]}},
{"name": "edit_file", "description": "Replace exact text in a file once.",
"input_schema": {"type": "object",
"properties": {"path": {"type": "string"},
"old_text": {"type": "string"},
"new_text": {"type": "string"}},
"required": ["path", "old_text", "new_text"]}},
{"name": "glob", "description": "Find files matching a glob pattern.",
"input_schema": {"type": "object",
"properties": {"pattern": {"type": "string"}},
"required": ["pattern"]}},
]
TOOL_HANDLERS = {
"bash": run_bash,
"read_file": run_read,
"write_file": run_write,
"edit_file": run_edit,
"glob": run_glob,
}
# -- From s04: hooks and permission checks --
HOOKS = {"UserPromptSubmit": [], "PreToolUse": [], "PostToolUse": [], "Stop": []}
def register_hook(event: str, callback):
HOOKS[event].append(callback)
def trigger_hooks(event: str, *args):
for callback in HOOKS[event]:
result = callback(*args)
if result is not None:
return result
return None
DENY_LIST = ["rm -rf /", "sudo", "shutdown", "reboot", "mkfs", "dd if="]
DESTRUCTIVE = ["rm ", "> /etc/", "chmod 777"]
def permission_hook(block):
if block.name == "bash":
command = block.input.get("command", "")
for pattern in DENY_LIST:
if pattern in command:
print(f"\n\033[31m[blocked] '{pattern}'\033[0m")
return "Permission denied by deny list"
if any(keyword in command for keyword in DESTRUCTIVE):
print("\n\033[33m[permission] Potentially destructive command\033[0m")
print(f" Tool: {block.name}({block.input})")
choice = input(" Allow? [y/N] ").strip().lower()
if choice not in ("y", "yes"):
return "Permission denied by user"
if block.name in ("read_file", "write_file", "edit_file"):
path = block.input.get("path", "")
if not (WORKDIR / path).resolve().is_relative_to(WORKDIR):
print("\n\033[33m[permission] Access outside workspace\033[0m")
print(f" Tool: {block.name}({block.input})")
choice = input(" Allow? [y/N] ").strip().lower()
if choice not in ("y", "yes"):
return "Permission denied by user"
return None
def log_hook(block):
preview = str(list(block.input.values())[:2])[:60]
print(f"\033[90m[HOOK] {block.name}({preview})\033[0m")
return None
def large_output_hook(block, output):
if len(str(output)) > 100000:
print(
f"\033[33m[HOOK] Large output from {block.name}: "
f"{len(str(output))} chars\033[0m"
)
return None
def context_inject_hook(query: str):
print(f"\033[90m[HOOK] UserPromptSubmit: working in {WORKDIR}\033[0m")
return None
def summary_hook(messages: list):
tool_count = sum(
1
for message in messages
for block in (
message.get("content")
if isinstance(message.get("content"), list)
else []
)
if isinstance(block, dict) and block.get("type") == "tool_result"
)
print(f"\033[90m[HOOK] Stop: session used {tool_count} tool calls\033[0m")
return None
register_hook("UserPromptSubmit", context_inject_hook)
register_hook("PreToolUse", permission_hook)
register_hook("PreToolUse", log_hook)
register_hook("PostToolUse", large_output_hook)
register_hook("Stop", summary_hook)
def call_tool(block) -> str:
handler = TOOL_HANDLERS.get(block.name)
try:
output = handler(**block.input) if handler else f"Unknown: {block.name}"
except Exception as error:
output = f"Error: {error}"
return str(output)
# -- New in s11: background execution --
class BackgroundManager:
def __init__(self):
self.tasks: dict[str, dict] = {}
self.results: dict[str, str] = {}
self._ready: list[str] = []
self._counter = 0
self._lock = threading.Lock()
def start(self, block) -> str:
if block.name != "bash":
raise ValueError("Only Bash commands can run in the background")
command = block.input.get("command")
if not isinstance(command, str) or not command.strip():
raise ValueError("Bash command cannot be empty")
with self._lock:
self._counter += 1
task_id = f"bg_{self._counter:04d}"
self.tasks[task_id] = {
"tool_use_id": block.id,
"command": command,
"status": "running",
}
thread = threading.Thread(
target=self._run,
args=(task_id, command),
daemon=True,
)
try:
thread.start()
except Exception:
with self._lock:
self.tasks.pop(task_id, None)
raise
print(f" [background] started {task_id}: {command[:60]}")
return task_id
def _run(self, task_id: str, command: str):
try:
output, exit_code = _run_bash_process(command)
result = _format_bash_result(output, exit_code)
status = "completed" if exit_code == 0 else "failed"
except Exception as error:
result = f"Error: {type(error).__name__}: {error}"
status = "failed"
with self._lock:
task = self.tasks.get(task_id)
if task is None:
return
task["status"] = status
self.results[task_id] = result
self._ready.append(task_id)
def collect(self) -> list[str]:
with self._lock:
ready = []
for task_id in self._ready:
task = self.tasks.pop(task_id, None)
result = self.results.pop(task_id, "")
if task is not None:
ready.append((task_id, task, result))
self._ready.clear()
notifications = []
for task_id, task, result in ready:
notifications.append(
f"<task_notification>\n"
f" <task_id>{task_id}</task_id>\n"
f" <status>{task['status']}</status>\n"
f" <command>{task['command']}</command>\n"
f" <summary>{result[:500]}</summary>\n"
f"</task_notification>"
)
print(f" [background] collected {task_id}: {task['status']}")
return notifications
BACKGROUND = BackgroundManager()
background_tasks = BACKGROUND.tasks
background_results = BACKGROUND.results
def should_run_background(tool_name: str, tool_input: dict) -> bool:
return (
tool_name == "bash"
and tool_input.get("run_in_background") is True
)
def start_background_task(block) -> str:
return BACKGROUND.start(block)
def collect_background_results() -> list[str]:
return BACKGROUND.collect()
def inject_background_results(messages: list) -> int:
notifications = collect_background_results()
if not notifications:
return 0
blocks = [{"type": "text", "text": item} for item in notifications]
if messages and messages[-1].get("role") == "user":
content = messages[-1].get("content", "")
if isinstance(content, list):
content.extend(blocks)
else:
messages[-1]["content"] = [
{"type": "text", "text": str(content)},
*blocks,
]
else:
messages.append({"role": "user", "content": blocks})
return len(notifications)
def execute_tool(block) -> str:
blocked = trigger_hooks("PreToolUse", block)
if blocked is not None:
return str(blocked)
if should_run_background(block.name, block.input):
try:
task_id = start_background_task(block)
output = (
f"[Background task {task_id} started] "
"The result will be collected on a later turn."
)
except Exception as error:
output = f"Error: {error}"
else:
output = call_tool(block)
trigger_hooks("PostToolUse", block, output)
return output
# -- Agent loop --
def agent_loop(messages: list):
while True:
inject_background_results(messages)
response = client.messages.create(
model=MODEL,
system=SYSTEM,
messages=messages,
tools=TOOLS,
max_tokens=8000,
)
messages.append({"role": "assistant", "content": response.content})
if response.stop_reason != "tool_use":
force = trigger_hooks("Stop", messages)
if force:
messages.append({"role": "user", "content": force})
continue
return
results = []
for block in response.content:
if block.type != "tool_use":
continue
output = execute_tool(block)
results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": output,
})
messages.append({"role": "user", "content": results})
if __name__ == "__main__":
print("s11: Background Tasks - explicit background Bash execution")
print("Enter a question, press Enter to send. Type q to quit.\n")
history = []
while True:
try:
query = input("\033[36ms11 >> \033[0m")
except (EOFError, KeyboardInterrupt):
break
if query.strip().lower() in ("q", "exit", ""):
break
trigger_hooks("UserPromptSubmit", query)
history.append({"role": "user", "content": query})
agent_loop(history)
for block in history[-1]["content"]:
if getattr(block, "type", None) == "text":
print(block.text)
print()
Try it — Background Tasks scenario
Long-running work moves to a thread while the agent keeps the main loop responsive.