mini agent

# Configurable settings.
# Use environment variables to change providers without editing code.

CONFIG = {
    # Any compatible Chat Completions endpoint.
    "base_url": os.environ.get("OPENAI_BASE_URL", "https://api.openai.com/v1"),
    "api_key": os.environ.get("OPENAI_API_KEY", ""),
    "model": os.environ.get("MODEL_ID", "gpt-4o-mini"),
    # Request parameters.
    "max_tokens": 8000,
    "request_timeout_s": 300,

    # Tool execution limits.
    "bash_timeout_s": 120,
    "max_output_chars": 50_000,

    # File reading.
    "max_read_lines": 2000,

    # Session persistence (chapter 5).
    # One JSONL file per session under the working directory by default.
    "session_dir": os.environ.get("MINI_AGENT_SESSION_DIR", ".mini_agent/sessions"),

    # Context compaction (chapter 6).
    # Approximate model context limit used to trigger summarization.
    "max_context_tokens": int(os.environ.get("MINI_AGENT_MAX_CONTEXT_TOKENS", "60000")),
    # Reserve room for the next request and response.
    "compaction_reserve_tokens": 8000,
    # Keep these recent user turns verbatim.
    "compaction_keep_recent_turns": 3,
}

# Literal shell-command blacklist.
# This demonstrates the idea of execution guardrails.
# Literal matching is easy to bypass; it is not a sandbox.
BLOCKED_PATTERNS = ["rm -rf /", "sudo", "shutdown", "reboot", "> /dev/"]


def require_api_key():
    """Require an API key before starting an agent session."""
    if not CONFIG["api_key"]:
        raise SystemExit(
            'Missing OPENAI_API_KEY.\nSet the environment variables, for example:\n  $env:OPENAI_BASE_URL="https://api.openai.com/v1"\n  $env:OPENAI_API_KEY="sk-..."\n  $env:MODEL_ID="gpt-4o-mini"\n'
        )

def call_llm(messages, tools, system=SYSTEM_PROMPT, signal=None):
    """Send one Chat Completions request and return the response dictionary.
    
    messages is the full conversation history; the model does not retain it
    between requests. tools contains function schemas. system=None uses the
    fallback prompt. signal performs cooperative cancellation checks."""
    # Chapter 4: check cancellation before sending.
    # Avoid sending a request after the user has already cancelled.
    if signal is not None:
        signal.check()

    # The protocol represents system instructions as a message.
    # Prepend it without changing the stored conversation history.
    system_text = SYSTEM_PROMPT if system is None else system
    full_messages = [{"role": "system", "content": system_text}] + messages

    payload = json.dumps(
        {
            "model": CONFIG["model"],
            "max_tokens": CONFIG["max_tokens"],
            "messages": full_messages,
            "tools": tools,
        }
    ).encode("utf-8")

    req = urllib.request.Request(
        f"{CONFIG['base_url']}/chat/completions",
        data=payload,
        method="POST",
        headers={
            "Content-Type": "application/json",
            "Authorization": f"Bearer {CONFIG['api_key']}",
        },
    )

    try:
        with urllib.request.urlopen(req, timeout=CONFIG["request_timeout_s"]) as resp:
            data = json.loads(resp.read().decode("utf-8"))
    except urllib.error.HTTPError as e:
        detail = e.read().decode("utf-8", errors="replace")[:500]
        raise RuntimeError('LLM request failed {0}: {1}'.format(e.code, detail)) from e
    except urllib.error.URLError as e:
        raise RuntimeError('LLM connection failed: {0}'.format(e.reason)) from e

    # Chapter 4: check cancellation again after receiving the response.
    # Discard the response if cancellation occurred while waiting.
    # urllib's blocking read cannot be interrupted by this check.
    # Immediate transport cancellation would require a different request mechanism.
    if signal is not None:
        signal.check()

    return data

def extract_message(response):
    """Return the complete assistant message, including content and tool_calls."""
    return response["choices"][0]["message"]


def extract_text(response):
    """Return assistant text, using an empty string when content is null."""
    return extract_message(response).get("content") or ""


def extract_tool_calls(response):
    """Return tool calls, or an empty list when the assistant requests none.
    
    The loop uses these calls to decide whether to continue, rather than
    depending on provider-specific finish_reason metadata."""
    return extract_message(response).get("tool_calls") or []

def tool_bash(inp):
    """Run a shell command with a literal blacklist, timeout, and output limit."""
    command = str(inp.get("command", ""))
    if any(pattern in command for pattern in BLOCKED_PATTERNS):
        return "Error: Dangerous command blocked"

    try:
        proc = subprocess.run(
            command,
            shell=True,
            cwd=os.getcwd(),
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
            timeout=CONFIG["bash_timeout_s"],
        )
    except subprocess.TimeoutExpired:
        return f"Error: command timed out after {CONFIG['bash_timeout_s']}s"

    output = f"{proc.stdout or ''}{proc.stderr or ''}".strip()
    if not output:
        return "(no output)"
    return output[: CONFIG["max_output_chars"]]

        # Step 5: ask the model.
        emit("request", turn=turn, history_len=len(messages))
        response = llm.call_llm(messages, tools, system=system, signal=signal)

        # Step 6: save the complete assistant message.
        # Keep tool_calls as well as content.
        # Their IDs pair with tool results in the next request.
        _append(messages, llm.extract_message(response), on_append)

        text = llm.extract_text(response)
        if text:
            emit("text", text=text)

        calls = llm.extract_tool_calls(response)

        # Step 7a: no tool calls means the task is done.
        if not calls:
            emit("done", turns=turn)
            break

        # Step 7b: check cancellation after the assistant declares tools.
        #
        # Do not simply break: the assistant call is already in history.
        # Each declared tool call must receive a matching result.
        # Otherwise the next request has invalid message ordering.
        # Providers may report an unmatched tool_call_id.
        #
        # Append an interrupted result for every pending call before exiting.
        # One call requires one matching tool message.
        if signal.aborted:
            for c in calls:
                _append(
                    messages,
                    {
                        "role": "tool",
                        "tool_call_id": c.get("id"),
                        "content": f"Error: aborted ({signal.reason})",
                    },
                    on_append,
                )
            emit("aborted", reason=signal.reason)
            break

        # Step 8: execute tools and append one message per result.
        # Multiple calls need separate tool messages with their own IDs.
        # Keep their order consistent with the declared tool calls.
        results = execute_tool_calls(calls, signal=signal, emit=emit, executor=executor)
        for r in results:
            _append(messages, {"role": "tool", **r}, on_append)

    return messages

def build_parser():
    p = argparse.ArgumentParser(
        prog="mini_agent",
        add_help=False,
        description='A minimal coding agent (educational project)',
    )
    p.add_argument("-h", "--help", action="help", help='Show this help message and exit')
    p.add_argument("-c", "--continue", dest="continue_", action="store_true",
                   help='Continue the most recent session')
    p.add_argument("--resume", metavar="SESSION_ID",
                   help='Resume a saved session')
    p.add_argument("--max-turns", type=int, default=None, metavar="N",
                   help='Maximum model calls per conversation turn')
    p.add_argument("-v", "--verbose", action="store_true",
                   help='Show tool output')
    p.add_argument("-p", "--print", dest="oneshot", metavar="PROMPT",
                   help='Run a task once and exit without entering interactive mode')
    return p


def main(argv=None):
    setup_console()
    args = build_parser().parse_args(argv)
    require_api_key()

    if args.oneshot:
        # One-shot mode runs the task and exits for script use.
        app = App(args)
        app.session.append("user", args.oneshot)
        agent_mod.agent_loop(
            app.session.messages,
            system=app.system_prompt,
            signal=app.signal,
            max_turns=args.max_turns,
            emit=app.render,
            executor=app.executor,
            on_append=app.session.persist,
            before_request=app.before_request,
        )
        return 0

    App(args).repl()
    return 0

def _function(name, description, properties, required):
    """Build an OpenAI-style function tool schema."""
    return {
        "type": "function",
        "function": {
            "name": name,
            "description": description,
            "parameters": {
                "type": "object",
                "properties": properties,
                "required": required,
            },
        },
    }


TOOLS = [
    _function(
        "bash",
        "Run a shell command.",
        {"command": {"type": "string"}},
        ["command"],
    ),
    _function(
        "read",
        "Read a file's text content.",
        {
            "path": {"type": "string"},
            "offset": {"type": "integer", "description": "1-indexed start line"},
            "limit": {"type": "integer", "description": "max lines to read"},
        },
        ["path"],
    ),
    _function(
        "write",
        "Write content to a file (overwrites the whole file).",
        {
            "path": {"type": "string"},
            "content": {"type": "string"},
        },
        ["path", "content"],
    ),
    _function(
        "edit",
        "Replace old_string with new_string in a file. "
        "old_string must appear exactly once in the file.",
        {
            "path": {"type": "string"},
            "old_string": {"type": "string"},
            "new_string": {"type": "string"},
        },
        ["path", "old_string", "new_string"],
    ),
]

def safe_path(raw):
    """Resolve a path and reject targets outside the working directory.
    
    realpath also resolves symbolic links. Raise ValueError for an outside path.
    This check applies to file tools; it does not sandbox shell commands."""
    workdir = os.path.realpath(os.getcwd())
    target = os.path.realpath(os.path.join(workdir, raw or "."))
    if target != workdir and not target.startswith(workdir + os.sep):
        raise ValueError(f"path outside workdir: {raw}")
    return target

def tool_bash(inp):
    """Run a shell command with a literal blacklist, timeout, and output limit."""
    command = str(inp.get("command", ""))
    if any(pattern in command for pattern in BLOCKED_PATTERNS):
        return "Error: Dangerous command blocked"

    try:
        proc = subprocess.run(
            command,
            shell=True,
            cwd=os.getcwd(),
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
            timeout=CONFIG["bash_timeout_s"],
        )
    except subprocess.TimeoutExpired:
        return f"Error: command timed out after {CONFIG['bash_timeout_s']}s"

    output = f"{proc.stdout or ''}{proc.stderr or ''}".strip()
    if not output:
        return "(no output)"
    return output[: CONFIG["max_output_chars"]]


def tool_read(inp):
    """Read numbered UTF-8 text with offset/limit pagination, without a shell."""
    try:
        abs_path = safe_path(str(inp.get("path", "")))
    except ValueError as e:
        return f"Error: {e}"

    try:
        with open(abs_path, "r", encoding="utf-8", errors="replace") as f:
            lines = f.readlines()
    except OSError as e:
        return f"Error: {e}"

    offset = int(inp.get("offset") or 1)  # 1-indexed
    limit = int(inp.get("limit") or CONFIG["max_read_lines"])
    start = max(0, offset - 1)
    selected = lines[start : start + limit]

    # Include line numbers so the model can reference exact locations.
    numbered = "".join(
        f"{start + i + 1:>6}\t{line}" for i, line in enumerate(selected)
    )
    remaining = len(lines) - (start + len(selected))
    if remaining > 0:
        numbered += f"\n... ({remaining} more lines)"
    return numbered or "(empty file)"

def tool_write(inp):
    """Overwrite a whole file under the shared per-path mutation lock."""
    path = str(inp.get("path", ""))
    content = str(inp.get("content", ""))
    try:
        abs_path = safe_path(path)
    except ValueError as e:
        return f"Error: {e}"

    def mutate():
        try:
            os.makedirs(os.path.dirname(abs_path) or ".", exist_ok=True)
            with open(abs_path, "w", encoding="utf-8", newline="") as f:
                f.write(content)
        except OSError as e:
            return f"Error: {e}"
        return f"Wrote {len(content)} bytes to {path}"

    return mutation_queue.run(abs_path, mutate)

def tool_edit(inp):
    """Replace one exact match under the shared per-path mutation lock.
    
    Read, count matches, and write within the same protected operation. Require
    exactly one old_string match to avoid changing an unintended occurrence."""
    path = str(inp.get("path", ""))
    old_string = str(inp.get("old_string", ""))
    new_string = str(inp.get("new_string", ""))

    if not old_string:
        return "Error: old_string cannot be empty"

    try:
        abs_path = safe_path(path)
    except ValueError as e:
        return f"Error: {e}"

    def mutate():
        try:
            with open(abs_path, "r", encoding="utf-8", errors="replace") as f:
                content = f.read()
        except OSError as e:
            return f"Error: {e}"

        count = content.count(old_string)
        if count == 0:
            return f"Error: old_string not found in {path}"
        if count > 1:
            return (
                f"Error: old_string appears {count} times in {path}, must be unique. "
                "Include more surrounding context to make it unique."
            )

        try:
            with open(abs_path, "w", encoding="utf-8", newline="") as f:
                f.write(content.replace(old_string, new_string, 1))
        except OSError as e:
            return f"Error: {e}"
        return f"Edited {path}"

    # Protect the entire read, match, and write operation.
    return mutation_queue.run(abs_path, mutate)

HANDLERS = {
    "bash": tool_bash,
    "read": tool_read,
    "write": tool_write,
    "edit": tool_edit,
}


def submit(name, inp):
    """Dispatch a tool by name and return its result as text.
    
    Return errors to the model so it can correct a call instead of crashing
    the agent. Unexpected handler exceptions are also converted to result text."""
    handler = HANDLERS.get(name)
    if handler is None:
        known = ", ".join(sorted(HANDLERS))
        return f"Error: unknown tool '{name}'. Available tools: {known}"
    try:
        return handler(inp)
    except Exception as e:  # Convert unexpected handler failures into model-visible errors.
        return f"Error: {type(e).__name__}: {e}"

def _run_one(call, signal, emit):
    """Parse one call's JSON arguments, execute it, and return its result.
    
    Convert cancellation and tool exceptions to result text so every recorded
    assistant tool call can still receive its matching tool message."""
    fn = call.get("function") or {}
    name = fn.get("name")
    raw_args = fn.get("arguments")
    if isinstance(raw_args, str):
        try:
            inp = json.loads(raw_args)
        except json.JSONDecodeError:
            inp = {}
    else:
        inp = raw_args or {}
    if not isinstance(inp, dict):
        inp = {}

    emit("tool_start", name=name, input=inp)

    try:
        signal.check()                      # Chapter 4: check cancellation before execution too.
        output = toolbox.submit(name, inp)
    except Aborted as e:
        # Convert cancellation to a tool result rather than propagating it here.
        # Every assistant tool call must have a corresponding tool message.
        # A missing result would make the next request invalid.
        output = f"Error: aborted ({e})"
    except Exception as e:                  # Convert unexpected tool failures to result text.
        output = f"Error: {type(e).__name__}: {e}"

    emit("tool_end", name=name, output=output)

    return {"tool_call_id": call.get("id"), "content": output}

class FileMutationQueue:
    """Serialize file mutations using a lock per normalized absolute path."""

    def __init__(self):
        # Normalized path key -> lock.
        self._locks = {}
        # Protect the lock registry itself.
        self._registry = threading.Lock()

    @staticmethod
    def _key_for(path):
        """Normalize existing or new file paths with abspath and realpath."""
        return os.path.realpath(os.path.abspath(path))

        key = self._key_for(path)

        # Serialize lock creation so same-path callers share one lock.
        with self._registry:
            lock = self._locks.get(key)
            if lock is None:
                lock = threading.Lock()
                self._locks[key] = lock

    def run(self, path, fn):
        """Run the no-argument mutation fn under the target path's lock."""
        key = self._key_for(path)

        # Serialize lock creation so same-path callers share one lock.
        with self._registry:
            lock = self._locks.get(key)
            if lock is None:
                lock = threading.Lock()
                self._locks[key] = lock

        # Serialize operations per path; different paths use different locks.
        with lock:
            return fn()

def tool_edit(inp):
    """Replace one exact match under the shared per-path mutation lock.
    
    Read, count matches, and write within the same protected operation. Require
    exactly one old_string match to avoid changing an unintended occurrence."""
    path = str(inp.get("path", ""))
    old_string = str(inp.get("old_string", ""))
    new_string = str(inp.get("new_string", ""))

    if not old_string:
        return "Error: old_string cannot be empty"

    try:
        abs_path = safe_path(path)
    except ValueError as e:
        return f"Error: {e}"

    def mutate():
        try:
            with open(abs_path, "r", encoding="utf-8", errors="replace") as f:
                content = f.read()
        except OSError as e:
            return f"Error: {e}"

        count = content.count(old_string)
        if count == 0:
            return f"Error: old_string not found in {path}"
        if count > 1:
            return (
                f"Error: old_string appears {count} times in {path}, must be unique. "
                "Include more surrounding context to make it unique."
            )

        try:
            with open(abs_path, "w", encoding="utf-8", newline="") as f:
                f.write(content.replace(old_string, new_string, 1))
        except OSError as e:
            return f"Error: {e}"
        return f"Edited {path}"

    # Protect the entire read, match, and write operation.
    return mutation_queue.run(abs_path, mutate)

def execute_tool_calls(calls, *, signal, emit, executor=None):
    """Execute tool calls and return result dictionaries in call order.
    
    Each result contains tool_call_id and content. The caller adds role='tool'.
    The executor allows concurrent work; results retain the original call order."""
    if not calls:
        return []

    # A single call or no executor: execute sequentially.
    if len(calls) == 1 or executor is None:
        return [_run_one(c, signal, emit) for c in calls]

    # Run concurrently, but collect results in the original call order.
    futures = [executor.submit(_run_one, c, signal, emit) for c in calls]
    return [f.result() for f in futures]

class Aborted(Exception):
    """Raise at cancellation checkpoints to unwind nested operations."""

    def __init__(self, reason="aborted"):
        super().__init__(reason)
        self.reason = reason


class Signal:
    """Share cancellation state across threads with threading.Event."""

    def __init__(self):
        self._event = threading.Event()
        self.reason = None

    def abort(self, reason="user aborted"):
        """Request cancellation; preserve the first recorded reason."""
        if not self._event.is_set():
            self.reason = reason
        self._event.set()

    @property
    def aborted(self):
        """Return whether cancellation has been requested, without raising."""
        return self._event.is_set()

    def check(self):
        """Raise Aborted if cancellation has been requested."""
        if self._event.is_set():
            raise Aborted(self.reason)

class SteeringQueue:
    """Store follow-up instructions until the next model request.
    
    The main thread pushes input while the agent thread drains the queue.
    A list preserves multiple instructions instead of overwriting older input."""

    def __init__(self):
        self._items = []
        self._lock = threading.Lock()

    def push(self, text):
        """Enqueue one instruction from the input thread."""
        text = (text or "").strip()
        if not text:
            return
        with self._lock:
            self._items.append(text)

    def drain(self):
        """Atomically take all instructions and clear the queue."""
        with self._lock:
            items, self._items = self._items, []
        return items

    turn = 0
    while True:
        # Step 1: cancellation check (chapter 4).
        if signal.aborted:
            emit("aborted", reason=signal.reason)
            break

        # Step 2: drain steering instructions (chapter 4).
        # Add follow-ups before asking the model so the next request includes them.
        if steering is not None:
            for text in steering.drain():
                _append(messages, {"role": "user", "content": text}, on_append)
                emit("steering", text=text)

        # Step 3: model-call limit (chapter 4).
        turn += 1
        if max_turns is not None and turn > max_turns:
            emit("max_turns", limit=max_turns)
            break

        # Step 7b: check cancellation after the assistant declares tools.
        #
        # Do not simply break: the assistant call is already in history.
        # Each declared tool call must receive a matching result.
        # Otherwise the next request has invalid message ordering.
        # Providers may report an unmatched tool_call_id.
        #
        # Append an interrupted result for every pending call before exiting.
        # One call requires one matching tool message.
        if signal.aborted:
            for c in calls:
                _append(
                    messages,
                    {
                        "role": "tool",
                        "tool_call_id": c.get("id"),
                        "content": f"Error: aborted ({signal.reason})",
                    },
                    on_append,
                )
            emit("aborted", reason=signal.reason)
            break

    emit("tool_start", name=name, input=inp)

    try:
        signal.check()                      # Chapter 4: check cancellation before execution too.
        output = toolbox.submit(name, inp)
    except Aborted as e:
        # Convert cancellation to a tool result rather than propagating it here.
        # Every assistant tool call must have a corresponding tool message.
        # A missing result would make the next request invalid.
        output = f"Error: aborted ({e})"
    except Exception as e:                  # Convert unexpected tool failures to result text.
        output = f"Error: {type(e).__name__}: {e}"

    emit("tool_end", name=name, output=output)

    return {"tool_call_id": call.get("id"), "content": output}

    # Avoid sending a request after the user has already cancelled.
    if signal is not None:
        signal.check()

    # The protocol represents system instructions as a message.
    # Prepend it without changing the stored conversation history.
    system_text = SYSTEM_PROMPT if system is None else system
    full_messages = [{"role": "system", "content": system_text}] + messages

    # Chapter 4: check cancellation again after receiving the response.
    # Discard the response if cancellation occurred while waiting.
    # urllib's blocking read cannot be interrupted by this check.
    # Immediate transport cancellation would require a different request mechanism.
    if signal is not None:
        signal.check()

    return data

    def repl(self):
        print('{0}mini_agent{1} {2}(/help for commands, Ctrl+C to interrupt, /exit to quit){3}'.format(BOLD, RESET, DIM, RESET))
        print('{0}Working directory: {1}{2}'.format(DIM, os.getcwd(), RESET))
        print('{0}Model: {1}   Session: {2}{3}'.format(DIM, CONFIG['model'], self.session.id, RESET))

        while True:
            try:
                user_text = input('\n{0}{1}You >{2} '.format(BOLD, CYAN, RESET)).strip()
            except (EOFError, KeyboardInterrupt):
                print()
                break

            if not user_text:
                continue
            if user_text.startswith("/"):
                try:
                    self.handle_command(user_text)
                except SystemExit:
                    break
                continue

            self.run_turn(user_text)

        self.executor.shutdown(wait=False)
        print('{0}Session saved: {1}{2}'.format(DIM, self.session.path, RESET))

    def run_turn(self, user_text):
        """Run user input in a background thread while accepting follow-ups."""
        self.signal.reset()
        self.session.append("user", user_text)

        done = threading.Event()

        def worker():
            try:
                agent_mod.agent_loop(
                    self.session.messages,
                    system=self.system_prompt,
                    signal=self.signal,
                    steering=self.steering,
                    max_turns=self.args.max_turns,
                    emit=self.render,
                    executor=self.executor,
                    on_append=self.session.persist,           # Chapter 5: persist only; the loop already appended in memory.
                    before_request=self.before_request,      # Chapter 6: compact before requests.
                )
            except Exception as e:
                print('{0}[Error] {1}: {2}{3}'.format(RED, type(e).__name__, e, RESET))
            finally:
                done.set()

        thread = threading.Thread(target=worker, daemon=True, name="agent")
        thread.start()

        # Keep reading input while the background agent runs.
        while not done.is_set():
            try:
                line = input('{0}(Running… add instructions, Ctrl+C to interrupt){1} '.format(DIM, RESET)).strip()
            except EOFError:
                self.signal.abort("stdin closed")
                break
            except KeyboardInterrupt:
                print('\n{0}Interrupting…{1}'.format(YELLOW, RESET))
                self.signal.abort("user pressed Ctrl+C")
                break

            if not line:
                continue
            if line.startswith("/"):
                self.handle_command(line)          # Handle slash commands immediately.
            else:
                self.steering.push(line)           # Queue other input for the next model request.
                print('{0}  ↳ Queued for the next model call{1}'.format(YELLOW, RESET))

        thread.join(timeout=1)

class Session:
    """A conversation with an in-memory message list and a JSONL file."""

    def __init__(self, session_id, messages, path):
        self.id = session_id
        self.messages = messages      # The same message list used by agent_loop.
        self.path = path

    # Construction.

    @classmethod
    def create(cls, session_id=None, session_dir=None):
        """Create a session, generating an ID when none is provided."""
        session_id = session_id or time.strftime("%Y%m%d-%H%M%S") + "-" + uuid.uuid4().hex[:6]
        directory = session_dir or CONFIG["session_dir"]
        os.makedirs(directory, exist_ok=True)
        path = os.path.join(directory, f"{session_id}.jsonl")
        return cls(session_id, [], path)

    @classmethod
    def load(cls, path):
        """Reload JSONL messages, warning and skipping lines with invalid JSON."""
        if not os.path.exists(path):
            raise FileNotFoundError('Session file not found: {0}'.format(path))
        messages = []
        skipped = 0
        with open(path, "r", encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                try:
                    messages.append(json.loads(line))
                except json.JSONDecodeError:
                    skipped += 1
        session_id = os.path.splitext(os.path.basename(path))[0]
        if skipped:
            print('[Warning] Skipped {0} corrupt lines'.format(skipped))
        return cls(session_id, messages, path)

    def append(self, role, content):
        """Append a role/content message to memory and disk, then return it."""
        msg = {"role": role, "content": content}
        self.messages.append(msg)
        self._write_line(msg)
        return msg

    def append_message(self, msg):
        """Append a complete message to memory and disk, then return it.
        
        Do not use this as the loop's on_append hook: the loop has already appended
        the message, so doing it here would duplicate the in-memory entry."""
        self.messages.append(msg)
        self._write_line(msg)
        return msg

    def persist(self, msg):
        """Save a message that the loop has already appended to memory.
        
        Use this as on_append: the loop changes memory; this method writes to disk."""
        self._write_line(msg)
        return msg

        # The loop and caller still hold references to this list object.
        # Rebinding would leave later appends on an outdated list.
        # Slice assignment keeps memory and persistence connected.
        self.messages[:] = messages
        tmp = self.path + ".tmp"
        with open(tmp, "w", encoding="utf-8") as f:
            for msg in messages:
                f.write(json.dumps(msg, ensure_ascii=False) + "\n")
        os.replace(tmp, self.path)      # Replace the file atomically after writing the temporary copy.

    # Session inventory.

    def stats(self):
        return {

def _append(messages, msg, on_append):
    messages.append(msg)
    if on_append is not None:
        on_append(msg)
    return msg

    files = [
        os.path.join(directory, n)
        for n in os.listdir(directory)
        if n.endswith(".jsonl")
    ]
    files.sort(key=lambda p: os.path.getmtime(p), reverse=True)
    return files


def latest_session(session_dir=None):
    """Return the most recently modified session path, or None."""
    files = list_sessions(session_dir)
    return files[0] if files else None

    def __init__(self, args):
        self.args = args
        self.session = self._open_session()
        self.signal = Signal()
        self.steering = SteeringQueue()
        self.executor = agent_mod.make_executor()
        self.render = make_renderer(args.verbose)
        self.skills = prompt.load_skills()
        self.system_prompt = prompt.build_system_prompt(self.skills)
        self.before_request = self._make_before_request()

    # Assembly.

    def _open_session(self):
        """Start a new session, continue the latest, or resume a selected ID."""
        if self.args.resume:
            path = self._find_session_path(self.args.resume)
            loaded = session_mod.Session.load(path)
            print('{0}Resumed session {1} ({2} messages){3}'.format(DIM, loaded.id, len(loaded.messages), RESET))
            return loaded

        if self.args.continue_:
            path = session_mod.latest_session()
            if path:
                loaded = session_mod.Session.load(path)
                print('{0}Continuing session {1} ({2} messages){3}'.format(DIM, loaded.id, len(loaded.messages), RESET))
                return loaded
            print('{0}No previous session; starting a new one{1}'.format(DIM, RESET))

        return session_mod.Session.create()

    def _find_session_path(self, session_id):
        directory = CONFIG["session_dir"]
        path = os.path.join(directory, f"{session_id}.jsonl")
        if os.path.exists(path):
            return path
        # Allow a session ID prefix as a convenience.
        for candidate in session_mod.list_sessions(directory):
            if os.path.basename(candidate).startswith(session_id):
                return candidate
        raise SystemExit('Session not found: {0} (use /list to see available sessions)'.format(session_id))

    def run_turn(self, user_text):
        """Run user input in a background thread while accepting follow-ups."""
        self.signal.reset()
        self.session.append("user", user_text)

        done = threading.Event()

        def worker():
            try:
                agent_mod.agent_loop(
                    self.session.messages,
                    system=self.system_prompt,
                    signal=self.signal,
                    steering=self.steering,
                    max_turns=self.args.max_turns,
                    emit=self.render,
                    executor=self.executor,
                    on_append=self.session.persist,           # Chapter 5: persist only; the loop already appended in memory.
                    before_request=self.before_request,      # Chapter 6: compact before requests.
                )
            except Exception as e:
                print('{0}[Error] {1}: {2}{3}'.format(RED, type(e).__name__, e, RESET))
            finally:
                done.set()

    def handle_command(self, line):
        parts = line.split(maxsplit=1)
        cmd = parts[0].lower()
        arg = parts[1].strip() if len(parts) > 1 else ""

        if cmd in ("/exit", "/quit", "/q"):
            raise SystemExit(0)
        if cmd == "/help":
            print(HELP_TEXT)
        elif cmd == "/new":
            self.session = session_mod.Session.create()
            print('{0}New session: {1}{2}'.format(DIM, self.session.id, RESET))
        elif cmd == "/list":
            files = session_mod.list_sessions()
            if not files:
                print('(No sessions yet)')
            for path in files[:20]:
                name = os.path.basename(path)
                size = os.path.getsize(path)
                print(f"  {name}  ({size} bytes)")
        elif cmd == "/resume":
            if not arg:
                print('Usage: /resume <session-id>')
            else:
                try:
                    self.session = session_mod.Session.load(self._find_session_path(arg))
                    print('{0}Switched to {1}{2}'.format(DIM, self.session.id, RESET))
                except SystemExit as e:
                    print(e)

def estimate_tokens(messages):
    """Estimate tokens from content, tool names, arguments, and tool-call IDs.
    
    Use characters / 3 plus fixed message overhead. This is a rough heuristic
    for mixed English and Chinese text, not an exact model-specific count."""
    total_chars = 0
    for msg in messages:
        content = msg.get("content")
        if isinstance(content, str):
            total_chars += len(content)
        elif content:                       # Fallback for uncommon historical content shapes.
            total_chars += len(str(content))
        for tc in msg.get("tool_calls") or []:
            fn = tc.get("function") or {}
            total_chars += len(str(fn.get("name", "")))
            total_chars += len(str(fn.get("arguments", "")))
        if msg.get("role") == "tool":
            total_chars += len(str(msg.get("tool_call_id", "")))
    # Allow fixed overhead for roles and message structure.
    overhead = len(messages) * 12
    return total_chars // 3 + overhead


def needs_compaction(messages, limit=None, reserve=None):
    """Decide whether estimated tokens have reached the reserved threshold."""
    limit = limit if limit is not None else CONFIG["max_context_tokens"]
    reserve = reserve if reserve is not None else CONFIG["compaction_reserve_tokens"]
    return estimate_tokens(messages) >= (limit - reserve)

def is_turn_start(msg):
    """Identify a user text message that starts a conversation turn."""

    if msg.get("role") != "user":
        return False
    return isinstance(msg.get("content"), str)