Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions src/lecode/agent/runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -304,6 +304,7 @@ async def run(
replay = self.store.load_for_model(self.session)
if replay and replay[0].get("role") == "system" and history[:1] == replay[:1]:
history.insert(0, {"role": "system", "content": ""})
del replay
# The live conversation, visible through ctx (subagents, hooks).
self.ctx.extras["conversation"] = history
manager = self.ctx.extras.get("workers")
Expand Down
71 changes: 45 additions & 26 deletions src/lecode/agent/tools/bash.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,12 +11,14 @@
from __future__ import annotations

import asyncio
import codecs
import contextlib
import tempfile
import time
import uuid
from collections.abc import Callable
from pathlib import Path
from typing import Any
from typing import Any, BinaryIO

from lecode.agent.tools.base import Tool, ToolContext, ToolResult
from lecode.extras.proc import ProcResult
Expand Down Expand Up @@ -71,22 +73,20 @@ async def _run_shell(
buffer = bytearray()
deadline = time.monotonic() + timeout
idle_deadline = time.monotonic() + idle_timeout
timed_out = idle_killed = False
timed_out = idle_killed = aborted = False

try:
while True:
wait = min(deadline, idle_deadline) - time.monotonic()
if wait <= 0:
timed_out = time.monotonic() >= deadline
idle_killed = not timed_out
_kill_tree(proc)
break
try:
chunk = await asyncio.wait_for(proc.stdout.read(65536), timeout=wait)
except TimeoutError:
timed_out = time.monotonic() >= deadline
idle_killed = not timed_out
_kill_tree(proc)
break
if not chunk: # EOF: process exited and pipes drained
break
Expand All @@ -96,27 +96,44 @@ async def _run_shell(
if len(buffer) > max_bytes:
del buffer[: len(buffer) - max_bytes]
idle_deadline = time.monotonic() + idle_timeout
except asyncio.CancelledError:
# Turn aborted (Ctrl-C): never leave the child running.
_kill_tree(proc)
with contextlib.suppress(TimeoutError, asyncio.CancelledError):
await asyncio.wait_for(proc.wait(), timeout=5.0)
except BaseException:
aborted = True
raise
finally:
if aborted or timed_out or idle_killed:
_kill_tree(proc)
with contextlib.suppress(TimeoutError, asyncio.CancelledError):
async with asyncio.timeout(5.0):
# wait() can wait on pipes held by children missed during a fork.
while proc.returncode is None:
await asyncio.sleep(0.01)
# Catch children spawned while the first signal killed the shell.
_kill_tree(proc)
while await proc.stdout.read(65536):
pass
await proc.wait()

with contextlib.suppress(TimeoutError):
await asyncio.wait_for(proc.wait(), timeout=5.0)
exit_code = proc.returncode if proc.returncode is not None else -1
return bytes(buffer), exit_code, timed_out, idle_killed


def _save_overflow(text: str) -> Path:
def _save_overflow(output: BinaryIO) -> tuple[Path, str, str]:
from lecode.config.loader import config_dir

overflow_dir = config_dir() / "overflow"
overflow_dir.mkdir(parents=True, exist_ok=True)
path = overflow_dir / f"{uuid.uuid4().hex}.log"
path.write_text(text, encoding="utf-8")
return path
head = tail = ""
half = MAX_OUTPUT_BYTES // 2
chunks = iter(lambda: output.read(65536), b"")
with path.open("w", encoding="utf-8") as destination:
for text in codecs.iterdecode(chunks, "utf-8", errors="replace"):
destination.write(text)
head += text[: half - len(head)]
tail = (tail + text)[-half:]
return path, head, tail


class BashTool(Tool):
Expand Down Expand Up @@ -185,22 +202,24 @@ async def run(self, args: dict, ctx: ToolContext) -> ToolResult:
return ToolResult(f"error: {e}", is_error=True)
return ToolResult(f"background task {record.id} started: {command}")
try:
output, exit_code, timed_out, idle_killed = await _run_shell(
command, ctx.cwd, timeout, idle, MAX_OUTPUT_BYTES
)
with tempfile.SpooledTemporaryFile(max_size=MAX_OUTPUT_BYTES) as output:
_, exit_code, timed_out, idle_killed = await _run_shell(
command, ctx.cwd, timeout, idle, MAX_OUTPUT_BYTES, on_chunk=output.write
)
truncated = output.tell() > MAX_OUTPUT_BYTES
output.seek(0)
if truncated:
full_path, head, tail = _save_overflow(output)
text = (
head
+ f"\n… [output truncated; full output saved to {full_path}] …\n"
+ tail
)
else:
text = output.read().decode("utf-8", errors="replace")
except OSError as e:
return ToolResult(f"error: {e}", is_error=True)

text = output.decode("utf-8", errors="replace")
if len(output) > MAX_OUTPUT_BYTES:
full_path = _save_overflow(text)
half = MAX_OUTPUT_BYTES // 2
text = (
text[:half]
+ f"\n… [output truncated; full output saved to {full_path}] …\n"
+ text[-half:]
)

notes: list[str] = []
if timed_out:
notes.append(f"timed out after {timeout}s")
Expand All @@ -218,7 +237,7 @@ async def run(self, args: dict, ctx: ToolContext) -> ToolResult:
stdout=text,
stderr="", # The shell merges stderr into stdout.
timed_out=timed_out or idle_killed,
truncated=len(output) > MAX_OUTPUT_BYTES,
truncated=truncated,
)
},
)
Expand Down
17 changes: 14 additions & 3 deletions src/lecode/agent/tools/read.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
from __future__ import annotations

import zlib
from itertools import islice
from pathlib import Path

from lecode.agent.tools.base import Tool, ToolContext, ToolResult
Expand Down Expand Up @@ -97,12 +98,22 @@ async def run(self, args: dict, ctx: ToolContext) -> ToolResult:
with_anchors = bool(args.get("with_anchors"))

try:
lines = path.read_text(encoding="utf-8", errors="replace").splitlines()
with path.open(encoding="utf-8", errors="replace") as stream:
lines = (text for line in stream for text in line.splitlines())
page: list[str] = []
total = 0
for total, text in enumerate(lines, start=1):
if offset <= total < offset + limit:
page.append(text)
if limit < 0:
# Preserve negative slice bounds without retaining the whole file.
start, stop, _ = slice(offset - 1, offset - 1 + limit).indices(total)
stream.seek(0)
lines = (text for line in stream for text in line.splitlines())
page = list(islice(lines, start, stop))
except OSError as e:
return ToolResult(f"error: {e}", is_error=True)

total = len(lines)
page = lines[offset - 1 : offset - 1 + limit]
with_marks: list[str] = []
for i, text in enumerate(page, start=offset):
if with_anchors:
Expand Down
13 changes: 9 additions & 4 deletions src/lecode/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@
import sys
from collections.abc import Callable, Coroutine
from pathlib import Path
from typing import Annotated, Any
from typing import TYPE_CHECKING, Annotated, Any

import typer
from typer.core import TyperGroup, TyperOption
Expand Down Expand Up @@ -63,10 +63,10 @@
SessionNotFoundError,
SessionStore,
)
from lecode.setup_wizard import offer_first_run_setup, run_wizard
from lecode.telemetry import init_telemetry, shutdown_telemetry
from lecode.tui.app import TuiApp
from lecode.tui.name_prompt import prompt_session_name

if TYPE_CHECKING:
from lecode.tui.app import TuiApp

#: Exit codes (headless mode uses the same taxonomy).
EXIT_OK = 0
Expand Down Expand Up @@ -688,13 +688,16 @@ def run_interactive(
"""
from rich.console import Console

from lecode.setup_wizard import offer_first_run_setup
from lecode.tui.app import TuiApp
from lecode.tui.loading import (
LoadingProgress,
build_load_report,
config_step,
provider_step,
session_step,
)
from lecode.tui.name_prompt import prompt_session_name

# Banner first — before any slow work (setup wizard, network fetches).
console = Console(no_color=no_color)
Expand Down Expand Up @@ -915,6 +918,8 @@ def run_setup() -> int:
if not sys.stdin.isatty():
typer.echo("error: --setup requires an interactive terminal", err=True)
return EXIT_STARTUP
from lecode.setup_wizard import run_wizard

try:
path = asyncio.run(run_wizard())
except (KeyboardInterrupt, EOFError):
Expand Down
74 changes: 52 additions & 22 deletions src/lecode/extras/proc.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,20 +36,27 @@ class ProcResult:
truncated: bool = False


def _cap(data: bytes, max_bytes: int) -> tuple[str, bool]:
"""Decode with head/tail keeping; the middle is elided past the cap."""
if len(data) <= max_bytes:
return data.decode("utf-8", errors="replace"), False
async def _read_capped(stream: asyncio.StreamReader, max_bytes: int) -> tuple[str, bool]:
"""Drain a pipe while retaining the reported prefix and suffix."""
head = bytearray()
tail = bytearray()
total = 0
half = max_bytes // 2
head = data[:half]
tail = data[-half:]
skipped = len(data) - len(head) - len(tail)
text = (
head.decode("utf-8", errors="replace")
+ TRUNCATION_MARKER.format(skipped=skipped)
+ tail.decode("utf-8", errors="replace")
while chunk := await stream.read(65536):
total += len(chunk)
head.extend(chunk[: max_bytes - len(head)])
if half:
tail.extend(chunk)
if len(tail) > half:
del tail[:-half]
if total <= max_bytes:
return bytes(head).decode("utf-8", errors="replace"), False
return (
bytes(head[:half]).decode("utf-8", errors="replace")
+ TRUNCATION_MARKER.format(skipped=total - half - len(tail))
+ bytes(tail).decode("utf-8", errors="replace"),
True,
)
return text, True


async def run_proc(
Expand All @@ -70,27 +77,50 @@ async def run_proc(
limit=max_output_bytes * 2,
start_new_session=True, # own process group so _kill works on trees
)

async def feed_input() -> None:
if proc.stdin is not None:
try:
proc.stdin.write(input.encode())
await proc.stdin.drain()
except (BrokenPipeError, ConnectionResetError):
pass # Match communicate(): a child may close stdin early.
finally:
proc.stdin.close()

tasks = [
asyncio.create_task(_read_capped(proc.stdout, max_output_bytes)),
asyncio.create_task(_read_capped(proc.stderr, max_output_bytes)),
asyncio.create_task(feed_input()),
]

async def communicate():
stdout, stderr, _ = await asyncio.gather(*tasks)
await proc.wait()
return stdout, stderr

communication = asyncio.create_task(communicate())
timed_out = False
try:
stdout_b, stderr_b = await asyncio.wait_for(
proc.communicate(input.encode() if input is not None else None),
timeout=timeout,
)
results = await asyncio.wait_for(asyncio.shield(communication), timeout=timeout)
except TimeoutError:
timed_out = True
_kill(proc)
try:
stdout_b, stderr_b = await asyncio.wait_for(proc.communicate(), timeout=5.0)
results = await asyncio.wait_for(asyncio.shield(communication), timeout=5.0)
except TimeoutError:
stdout_b, stderr_b = b"", b""
except asyncio.CancelledError:
# Caller aborted (Ctrl-C): never leave the child running.
results = [("", False), ("", False)]
except BaseException:
_kill(proc)
with contextlib.suppress(TimeoutError, asyncio.CancelledError):
await asyncio.wait_for(proc.wait(), timeout=5.0)
raise
stdout, out_truncated = _cap(stdout_b or b"", max_output_bytes)
stderr, err_truncated = _cap(stderr_b or b"", max_output_bytes)
finally:
communication.cancel()
for task in tasks:
task.cancel()
await asyncio.gather(communication, *tasks, return_exceptions=True)
(stdout, out_truncated), (stderr, err_truncated) = results[:2]
exit_code = proc.returncode if proc.returncode is not None else -1
return ProcResult(
exit_code=exit_code,
Expand Down
Loading
Loading