A primitive agent harness written to describe what an agent harness is.

ask-model.py

"""Stream a prompt to an Ollama model, printing its answer.

The answer goes to stdout and everything else to stderr, so redirecting stdout
captures the answer alone:

    python ask-model.py "why is the sky blue?" > answer.txt

Reasoning is off by default; --think low|medium|high turns it on, and the
reasoning then streams to stderr while the answer still goes to stdout.

--file passes a file to the model as context ahead of the prompt:

    python ask-model.py --file notes.md "summarise this"

--tools lets the model work the files for itself instead, by calling read_file
and write_file when the prompt gives it reason to. It is off by default, every
call it makes is put to you for approval, and write_file will only write inside
the working directory:

    python ask-model.py --tools "summarise notes.md into summary.md"

Each call and its result are reported on stderr, and the model may make several
rounds of them before answering; --max-tool-turns caps how many.

A gate stands in front of those calls, and by default every one of them is put
to you at the terminal before it runs. --approve writes narrows the asking to
the tools that change something, and --approve off runs them unasked. Refusing
a call tells the model so rather than ending the run:

    python ask-model.py --tools --approve writes "tidy up notes.md"

Because the gate has to ask somebody, --tools needs a terminal unless it is
turned off:

    python ask-model.py --tools --approve off "summarise notes.md" > out.txt

--memory records the session to a JSON file: the prompts, the answers, the
reasoning, the tool calls and their results — everything the model was sent and
everything it sent back. The file is rewritten after every message, so an
interrupted run still leaves a record of what happened:

    python ask-model.py --memory session.json --tools "what is in notes.md?"

--load-memory starts from a saved session instead of an empty one, so a later
run can be asked about an earlier one:

    python ask-model.py --load-memory session.json "what was the passphrase?"

Pass both, naming the same file, to continue a session and keep recording it.
Nothing is written unless --memory names a file.

At a terminal the answer is rendered as markdown as it streams. When stdout is
redirected the raw markdown is written through untouched, so the file stays a
valid markdown document.

Rendering needs rich (pip install rich); without it, or with --no-markdown, the
answer prints as plain text.
"""

import argparse
import datetime
import json
import os
import sys

from ollama import Client

try:
  from rich.console import Console
  from rich.live import Live
  from rich.markdown import Markdown
except ImportError:  # Optional: without rich the answer prints as plain text.
  Console = Live = Markdown = None

DIM = '\033[2m'
BOLD = '\033[1m'
RESET = '\033[0m'

# A tool is four parts, and only the last of them is code: the name the model
# calls, the description telling it when to, the schema shaping its arguments,
# and the handler below that actually runs. The first three are sent to the
# model on every request, which is why they are worth writing carefully and
# worth keeping few.
TOOLS = [
  {
    'type': 'function',
    'function': {
      'name': 'read_file',
      'description': (
        'Read a UTF-8 text file and return its contents. Use this whenever the '
        'question refers to a file you have not been shown.'
      ),
      'parameters': {
        'type': 'object',
        'properties': {
          'path': {
            'type': 'string',
            'description': 'Path to the file, relative to the working directory.',
          },
        },
        'required': ['path'],
      },
    },
  },
  {
    'type': 'function',
    'function': {
      'name': 'write_file',
      'description': (
        'Write a UTF-8 text file, creating it or replacing whatever is already '
        'there. The path must stay inside the working directory: paths above it '
        'or on another drive are refused.'
      ),
      'parameters': {
        'type': 'object',
        'properties': {
          'path': {
            'type': 'string',
            'description': 'Path to the file, relative to the working directory.',
          },
          'content': {
            'type': 'string',
            'description': 'The complete text to write. It replaces the whole file.',
          },
        },
        'required': ['path', 'content'],
      },
    },
  },
]

# A tool result is re-sent with every later request in the conversation, so an
# unbounded one is paid for repeatedly. Cap it, and say so where the model can
# see the cap, rather than letting one large file crowd out everything else.
MAX_RESULT_CHARS = 20_000

# Stamped into every file written and checked on every file read, so a session
# saved by an older script is refused rather than half-understood.
MEMORY_VERSION = 1


def styles(stream):
  """ANSI codes for a stream, or empty strings when it is not a terminal."""
  if stream.isatty():
    return DIM, BOLD, RESET
  return '', '', ''


def read_file(path):
  """Handler for the read_file tool: the contents, or why there are none."""
  try:
    # Read as UTF-8 rather than the Windows codepage, which mangles or rejects
    # anything the file writes outside ASCII.
    with open(path, encoding='utf-8') as handle:
      contents = handle.read()
  except OSError as error:
    return f'cannot read {path}: {error.strerror}'
  except UnicodeDecodeError:
    return f'cannot read {path}: not UTF-8 text'
  if len(contents) > MAX_RESULT_CHARS:
    return contents[:MAX_RESULT_CHARS] + f'\n[truncated at {MAX_RESULT_CHARS} characters]'
  return contents


def in_working_folder(path):
  """The absolute path to write to, or None if it lands outside the folder.

  A tool the model can aim anywhere is a tool that can overwrite anything the
  user owns. Reading is bounded by what is already there; writing is not, so
  this one is confined — and the confinement is stated in the tool description
  as well, since a rule the model cannot read costs it a round to discover.
  """
  # Real paths on both sides, so a symlink inside the folder cannot be used to
  # step outside it.
  working = os.path.realpath(os.getcwd())
  target = os.path.realpath(os.path.join(working, path))
  try:
    relative = os.path.relpath(target, working)
  except ValueError:  # a different drive on Windows has no relative path
    return None
  if relative == os.curdir or relative == os.pardir:
    return None
  if relative.startswith(os.pardir + os.sep):
    return None
  return target


def write_file(path, content):
  """Handler for the write_file tool: what was written, or why it was not."""
  target = in_working_folder(path)
  if target is None:
    return f'cannot write {path}: outside the working directory'
  # Checked before opening, because opening for writing truncates: by the time
  # the file is open there is no way to tell whether anything was lost.
  existed = os.path.exists(target)
  try:
    os.makedirs(os.path.dirname(target), exist_ok=True)
    # newline='' writes the content exactly as the model sent it. Without it,
    # Windows turns every \n into \r\n, so a file read by one tool and written
    # back by the other would come out with different bytes than it went in.
    with open(target, 'w', encoding='utf-8', newline='') as handle:
      handle.write(content)
  except OSError as error:
    return f'cannot write {path}: {error.strerror}'
  verb = 'overwrote' if existed else 'wrote'
  return f'{verb} {path}, {len(content):,} characters'


# The name the model calls, mapped to the code that runs. Everything above this
# line is written for the model to read; everything below it is not.
HANDLERS = {
  'read_file': read_file,
  'write_file': write_file,
}

# The tools that change something, and so the ones --approve writes asks about.
# Reading is bounded by what is already there; writing is not.
WRITING_TOOLS = {'write_file'}


def call_tool(name, arguments):
  """Dispatch one tool call, returning what the model should see as its result.

  Failures are results too: a model told that its call failed can correct
  itself, whereas an exception would end the run.
  """
  # Some servers render tool names into a syntax where a hyphen cannot survive,
  # so accept either spelling rather than depending on which one comes back.
  handler = HANDLERS.get(name.replace('-', '_'))
  if handler is None:
    return f'no such tool: {name}'
  try:
    return handler(**arguments)
  except TypeError as error:
    # The model invented arguments the handler does not take. Python's own
    # message names the ones it wanted, which is what the model needs to read.
    return f'cannot call {name}: {error}'


def describe(arguments):
  """A call's arguments on one line, with anything long summarised.

  write_file passes a whole file as an argument, so the arguments cannot simply
  be printed: the notice is there to make the call legible, not to repeat it.
  """
  parts = []
  for key, value in arguments.items():
    text = str(value)
    if len(text) > 40 or '\n' in text:
      text = f'<{len(text):,} chars>'
    parts.append(f'{key}={text}')
  return ' '.join(parts)


def preview(text, limit=5):
  """The start of a long argument, indented for the approval prompt.

  Approving a write without seeing what is being written is a formality rather
  than a decision, so the gate shows the content — but only the start of it, or
  a large file would scroll the question itself off the screen.
  """
  lines = text.splitlines() or ['']
  shown = [f'    {line[:76]}{"..." if len(line) > 76 else ""}' for line in lines[:limit]]
  if len(lines) > limit:
    shown.append(f'    ... {len(lines) - limit:,} more lines')
  return '\n'.join(shown)


def ask_approval(name, arguments):
  """Put one proposed call to the operator. Returns 'yes', 'no', 'all' or 'nobody'.

  This is the gate, and it sits here — between the model deciding and the
  handler running — for the same reason a lock goes on the door rather than in
  the visitor's instructions. It never reads the conversation, so nothing the
  model writes can talk it round.
  """
  dim, bold, reset = styles(sys.stderr)
  print(f'\n{bold}{name}{reset}{dim} {describe(arguments)}{reset}', file=sys.stderr)
  for value in arguments.values():
    text = str(value)
    if len(text) > 40 or '\n' in text:
      print(f'{dim}{preview(text)}{reset}', file=sys.stderr)

  while True:
    # The question goes to stderr and the reply is read from stdin, so the
    # answer on stdout stays the answer even while the operator is being asked.
    print(f'{dim}run it? [y]es / [N]o / [a]ll: {reset}', end='', flush=True, file=sys.stderr)
    try:
      reply = input().strip().lower()
    except EOFError:  # nobody there to ask, which is not a yes
      print(file=sys.stderr)
      return 'nobody'
    if reply in ('y', 'yes'):
      return 'yes'
    if reply in ('n', 'no', ''):  # a bare Enter is the safe answer, not the eager one
      return 'no'
    if reply in ('a', 'all'):
      return 'all'


def save_memory(path, model, tools, messages):
  """Write the whole session to disk, replacing the previous write atomically.

  The messages are the session: every prompt, every answer, every tool call and
  every result, in the order they happened. That list is also exactly what is
  sent to the model on the next request, so saving it saves the context itself
  rather than a summary of it.

  This is called after every message rather than once at the end, because a run
  that is written up at the end leaves nothing behind when it is the run that
  hung. Writing a temporary file alongside the target and renaming it means an
  interrupted write cannot leave a half-finished file where the session was.
  """
  memory = {
    'version': MEMORY_VERSION,
    'saved': datetime.datetime.now(datetime.timezone.utc).isoformat(timespec='seconds'),
    # The run that last wrote the session. A session continued with a different
    # model says so on stderr as it loads, and the messages keep their order
    # either way, so this names the last writer rather than every one of them.
    'model': model,
    # Recorded because the tools were part of the prompt too, not just the
    # messages: the model's answers only make sense against what it was offered.
    'tools': tools,
    'messages': messages,
  }
  temporary = path + '.tmp'
  # In the same directory as the target: a rename is only atomic within one
  # volume, so a temporary file elsewhere would silently lose the guarantee.
  with open(temporary, 'w', encoding='utf-8') as handle:
    json.dump(memory, handle, indent=2, ensure_ascii=False)
  os.replace(temporary, path)


def load_memory(path):
  """Read a saved session: its messages, its model, and the tools it was given."""
  with open(path, encoding='utf-8') as handle:
    memory = json.load(handle)  # raises ValueError if the file is not JSON
  if not isinstance(memory, dict) or not isinstance(memory.get('messages'), list):
    raise ValueError('not a session file')
  if memory.get('version') != MEMORY_VERSION:
    raise ValueError(f'unsupported session version {memory.get("version")!r}')

  messages = memory['messages']
  # A run interrupted between a tool call and its result ends on an assistant
  # turn whose calls were never answered. The record is right to keep it — it
  # is what happened — but a conversation cannot be resumed from it, so the
  # repair belongs here, on the way in, rather than in the file.
  if messages and messages[-1].get('role') == 'assistant' and messages[-1].get('tool_calls'):
    messages.pop()
  return messages, memory.get('model'), memory.get('tools') or []


def stream_turn(client, model, messages, think, tools, show_reasoning, render, console):
  """Stream one assistant turn, returning its reasoning, text and tool calls."""
  # Labels and reasoning are decoration around the answer, so they share stderr
  # and are styled for stderr; stdout carries the answer alone.
  dim, bold, reset = styles(sys.stderr)

  stream = client.chat(model, messages=messages, stream=True, think=think, tools=tools)

  in_reasoning = False
  live = None
  reasoning = []
  answer = []
  tool_calls = []

  try:
    for part in stream:
      message = part['message']

      # An unused field arrives as '' or None depending on the server build, so
      # test each one for truthiness before printing it.
      thinking = message.get('thinking')
      if thinking:
        reasoning.append(thinking)
        if show_reasoning:
          if not in_reasoning:
            print(f'{bold}Reasoning{reset}', file=sys.stderr)
            in_reasoning = True
          print(f'{dim}{thinking}{reset}', end='', flush=True, file=sys.stderr)

      content = message.get('content')
      if content:
        if not answer:
          if in_reasoning:
            print(f'\n{bold}Answer{reset}', file=sys.stderr)
          if render:
            # Started here rather than before the loop so its repainting never
            # interleaves with reasoning still streaming to stderr. The live
            # region is a transient preview: cropped to the window so rich
            # keeps its repaint anchor, then erased and replaced below by one
            # clean render of the whole answer.
            live = Live(
              console=console,
              transient=True,
              vertical_overflow='ellipsis',  # crop rather than lose the anchor
              refresh_per_second=12,
            )
            live.start()
        answer.append(content)
        if live:
          live.update(Markdown(''.join(answer)))
        else:
          print(content, end='', flush=True)

      # Tool calls arrive as structure rather than text, so they never reach
      # stdout: they are an instruction to the harness, not part of the answer.
      calls = message.get('tool_calls')
      if calls:
        tool_calls.extend(calls)
  finally:
    # Also runs on Ctrl-C, so the terminal is restored and a partial answer is
    # still printed in full rather than vanishing with the transient region.
    if live:
      live.stop()
      console.print(Markdown(''.join(answer)))

  # Leave both streams on a fresh line for the shell prompt. Rich ends its own
  # output with one, so only the plain path needs this.
  if in_reasoning and not answer:
    print(file=sys.stderr)
  if answer and not live:
    print()

  return ''.join(reasoning), ''.join(answer), tool_calls


def main():
  parser = argparse.ArgumentParser(
    description=__doc__,
    formatter_class=argparse.RawDescriptionHelpFormatter,
  )
  parser.add_argument('prompt', nargs='+', help='the prompt to send')
  parser.add_argument('--model', default='gpt-oss:120b-cloud', help='model to ask')
  parser.add_argument(
    '--file',
    metavar='PATH',
    help='a file to send as context ahead of the prompt',
  )
  parser.add_argument(
    '--tools',
    action='store_true',
    help='let the model call the read_file tool for itself (off by default)',
  )
  parser.add_argument(
    '--max-tool-turns',
    type=int,
    default=5,
    metavar='N',
    help='rounds of tool calls to allow before the model must answer',
  )
  parser.add_argument(
    '--approve',
    choices=['off', 'writes', 'all'],
    default='all',
    help="ask before running a tool: 'all' (the default) asks about every call, "
    "'writes' only about the ones that change something, 'off' runs them unasked",
  )
  parser.add_argument(
    '--memory',
    metavar='PATH',
    help='record the session to a JSON file, rewritten after every message',
  )
  parser.add_argument(
    '--load-memory',
    metavar='PATH',
    help='start from a session saved earlier instead of an empty one',
  )
  parser.add_argument(
    '--think',
    choices=['off', 'low', 'medium', 'high'],
    default='off',
    help="reasoning effort; 'off' (the default) disables reasoning",
  )
  parser.add_argument(
    '--hide-reasoning',
    action='store_true',
    help='request reasoning but do not print it',
  )
  parser.add_argument(
    '--no-markdown',
    action='store_true',
    help='print the raw markdown instead of rendering it',
  )
  args = parser.parse_args()

  # Zero would withdraw the tools before the model has been asked anything, and
  # a negative would skip the conversation entirely.
  if args.max_tool_turns < 1:
    parser.error('--max-tool-turns must be at least 1')

  # Windows defaults these to a legacy codepage, which mangles anything the
  # model writes outside ASCII (curly quotes, em dashes, accents).
  for stream in (sys.stdout, sys.stderr):
    if hasattr(stream, 'reconfigure'):
      stream.reconfigure(encoding='utf-8')

  think = False if args.think == 'off' else args.think
  show_reasoning = bool(think) and not args.hide_reasoning

  # Rendering markdown means emitting ANSI and repainting, so it is only safe
  # at a terminal; a redirect or a pipe gets the markdown source instead.
  render = Markdown is not None and not args.no_markdown and sys.stdout.isatty()

  dim, bold, reset = styles(sys.stderr)

  prompt = ' '.join(args.prompt)
  if args.file:
    # Fenced and labelled so the model can tell the context from the
    # instruction, which follows it: models attend best to the last thing said.
    try:
      # Read as UTF-8 rather than the Windows codepage, which mangles or
      # rejects anything the file writes outside ASCII.
      with open(args.file, encoding='utf-8') as handle:
        context = handle.read()
    except OSError as error:
      parser.error(f'cannot read {args.file}: {error.strerror}')
    except UnicodeDecodeError:
      parser.error(f'cannot read {args.file}: not UTF-8 text')
    prompt = f'{args.file}:\n```\n{context}\n```\n\n{prompt}'

  # Refused up front rather than denying every call in silence: a gate nobody
  # is there to answer is not a gate, it is a broken run waiting to happen.
  # Only when there are tools to gate, so a plain question still answers into a
  # pipe, and running unattended stays possible but has to be asked for.
  if args.tools and args.approve != 'off' and not sys.stdin.isatty():
    parser.error(
      f'--approve {args.approve} needs a terminal to ask at; '
      'pass --approve off to run the tools unattended'
    )

  if args.memory:
    # Checked before the model is asked anything, so a mistyped path costs a
    # message rather than a whole answer that then has nowhere to be written.
    directory = os.path.dirname(args.memory)
    if directory and not os.path.isdir(directory):
      parser.error(f'cannot write {args.memory}: no such directory {directory}')

  messages = []
  saved_tools = []
  if args.load_memory:
    try:
      messages, saved_model, saved_tools = load_memory(args.load_memory)
    except OSError as error:
      parser.error(f'cannot load {args.load_memory}: {error.strerror}')
    except ValueError as error:
      parser.error(f'cannot load {args.load_memory}: {error}')
    # Worth saying out loud: a transcript carries the previous model's habits,
    # its reasoning included, into a model that may not share them.
    if saved_model and saved_model != args.model:
      print(
        f'{dim}[memory recorded with {saved_model}, continuing with {args.model}]{reset}',
        file=sys.stderr,
      )

  # Every tool offered at some point in the session, rather than the ones this
  # run happens to offer: a resume without --tools would otherwise leave calls
  # in the messages with nothing to say what the model had been given, and a
  # resume with --tools would claim tools that earlier turns never saw.
  recorded_tools = list(saved_tools)
  if args.tools:
    recorded = {tool['function']['name'] for tool in recorded_tools}
    recorded_tools += [t for t in TOOLS if t['function']['name'] not in recorded]

  def remember(message):
    """Add a message to the context, and to the record if one is being kept."""
    messages.append(message)
    if args.memory:
      save_memory(args.memory, args.model, recorded_tools, messages)

  remember({'role': 'user', 'content': prompt})

  client = Client()
  console = Console(file=sys.stdout) if render else None

  # Set once the operator answers 'all', and never unset: being asked the same
  # question repeatedly is how a gate gets turned off for good.
  approved_everything = False
  # Set if the gate is answered by nobody, so the run says so once rather than
  # refusing call after call in silence.
  nobody_to_ask = False

  # The loop that makes this a harness rather than a chat client: ask, run
  # whatever the model asked for, tell it what happened, ask again. Without
  # --tools there is nothing for it to call, so the first turn is the only one.
  for turn in range(args.max_tool_turns + 1):
    if args.tools and turn == args.max_tool_turns:
      # The cap binds. Withdrawing the tools is not enough on its own: a model
      # that has been calling them will happily call one it was not offered,
      # and then the run would end with no answer at all. So say it in words
      # too, in the one place the model is certain to read — the last message.
      print(f'{dim}[tool limit reached; asking for an answer]{reset}', file=sys.stderr)
      remember({
        'role': 'user',
        'content': 'No more tool calls are available. Answer now with what you have.',
      })
    tools = TOOLS if args.tools and turn < args.max_tool_turns else None

    reasoning, answer, calls = stream_turn(
      client, args.model, messages, think, tools, show_reasoning, render, console
    )
    # The model's own turn goes back verbatim, reasoning included: it is the
    # context in which the results are about to arrive. The calls are copied
    # into plain dicts so that the conversation stays something that can be
    # written to a file and read back as it was.
    entry = {'role': 'assistant', 'content': answer}
    if reasoning:
      entry['thinking'] = reasoning
    # Calls the model made on a turn where it had no tools are left out: they
    # were never run, and an unanswered call cannot be resumed from.
    if calls and tools is not None:
      entry['tool_calls'] = [
        {
          'function': {
            'name': call['function']['name'],
            'arguments': dict(call['function']['arguments']),
          }
        }
        for call in calls
      ]
    remember(entry)

    # Nothing left to run, so whatever came back is the answer.
    if 'tool_calls' not in entry:
      break

    for call in entry['tool_calls']:
      name = call['function']['name']
      arguments = call['function']['arguments']
      # The gate: between the model deciding and the handler running, which is
      # the only place it can be. Before the model call it would not know what
      # was being proposed; after the handler, the file would already be gone.
      gated = args.approve == 'all' or (args.approve == 'writes' and name in WRITING_TOOLS)
      allowed = True
      if gated and not approved_everything:
        if nobody_to_ask:
          allowed = False
        else:
          decision = ask_approval(name, arguments)
          if decision == 'nobody':
            # Where an unattended run is really discovered: on Windows stdin
            # can claim to be a terminal when it is the null device, so the
            # check at startup lets one through that has nobody behind it.
            print(
              f'{dim}[nothing answered the gate, so the rest of this run is refused; '
              f'pass --approve off to run the tools unattended]{reset}',
              file=sys.stderr,
            )
            nobody_to_ask = True
          approved_everything = decision == 'all'
          allowed = decision in ('yes', 'all')

      # A refusal is information, not a crash. The model is told in the same
      # way it is told anything else, and can explain itself or try something
      # the operator might say yes to. It is told plainly that nothing
      # happened, because a model given only 'refused' will sometimes go on to
      # report the work as done.
      refusal = 'the user refused this call, so nothing happened. Say so; do not report it as done.'
      result = call_tool(name, arguments) if allowed else refusal

      # A short result is worth showing outright — it is usually a confirmation
      # or a refusal, and reading name, arguments and outcome on one line is
      # the only view of the loop the operator gets.
      if not allowed:
        outcome = 'refused'
      elif len(result) <= 60 and '\n' not in result:
        outcome = result
      else:
        outcome = f'{len(result):,} chars'
      print(
        f'{dim}[{reset}{bold}{name}{reset}{dim} {describe(arguments)} -> {outcome}]{reset}',
        file=sys.stderr,
      )
      # The result is just another message. The model has no other way to learn
      # what happened, so an empty or failed result must still be sent back.
      remember({'role': 'tool', 'tool_name': name, 'content': result})


if __name__ == '__main__':
  try:
    main()
  except KeyboardInterrupt:
    sys.exit(130)