Return the current snapshot of a background Bash task, or ``None``. Result shape mirrors what ``TaskOutput`` exposes to the model: { "task_id": ..., "status": "running" | "completed" | "failed", "exit_code": int | None, "command": str,
(
context: ToolContext,
task_id: str,
*,
max_bytes: int = 200_000,
)
| 177 | |
| 178 | |
| 179 | def read_background_output( |
| 180 | context: ToolContext, |
| 181 | task_id: str, |
| 182 | *, |
| 183 | max_bytes: int = 200_000, |
| 184 | ) -> dict[str, Any] | None: |
| 185 | """Return the current snapshot of a background Bash task, or ``None``. |
| 186 | |
| 187 | Result shape mirrors what ``TaskOutput`` exposes to the model: |
| 188 | { |
| 189 | "task_id": ..., |
| 190 | "status": "running" | "completed" | "failed", |
| 191 | "exit_code": int | None, |
| 192 | "command": str, |
| 193 | "output": str, # combined stdout+stderr, possibly truncated |
| 194 | "truncated": bool, # True if the log was bigger than ``max_bytes`` |
| 195 | "pid": int, |
| 196 | "started_at": float, |
| 197 | "finished_at": float | None, |
| 198 | } |
| 199 | """ |
| 200 | entry = context.background_bash_tasks.get(task_id) |
| 201 | if entry is None: |
| 202 | return None |
| 203 | |
| 204 | output_path = Path(entry["output_path"]) |
| 205 | try: |
| 206 | total_size = output_path.stat().st_size |
| 207 | except OSError: |
| 208 | total_size = 0 |
| 209 | |
| 210 | output_bytes = b"" |
| 211 | truncated = False |
| 212 | try: |
| 213 | with open(output_path, "rb") as fh: |
| 214 | if total_size > max_bytes: |
| 215 | fh.seek(total_size - max_bytes) |
| 216 | truncated = True |
| 217 | output_bytes = fh.read() |
| 218 | except OSError: |
| 219 | output_bytes = b"" |
| 220 | |
| 221 | try: |
| 222 | output_text = output_bytes.decode("utf-8", errors="replace") |
| 223 | except Exception: # pragma: no cover - decode("replace") shouldn't raise |
| 224 | output_text = "" |
| 225 | |
| 226 | exit_code = entry.get("exit_code") |
| 227 | # Strip the trailing __CLAWCODEX_EXIT__ marker we emit from the wrapper so |
| 228 | # it never leaks into the model's transcript. |
| 229 | marker = "__CLAWCODEX_EXIT__=" |
| 230 | if marker in output_text: |
| 231 | idx = output_text.rfind(marker) |
| 232 | # Trim everything from the last newline before the marker onward. |
| 233 | nl = output_text.rfind("\n", 0, idx) |
| 234 | if nl != -1: |
| 235 | output_text = output_text[:nl] |
| 236 | else: |