mini agent-2

CONFIG.update({
    "request_max_retries": int(os.environ.get("MINI_AGENT_REQUEST_MAX_RETRIES", "2")),
    "request_retry_base_s": 0.5,
    "request_retry_cap_s": 10.0,
})

class RequestFailure(RuntimeError):
    def __init__(self, category, *, status=None, retry_after=None):
        self.category = category
        self.status = status
        self.retry_after = retry_after
        labels = {
            "http_error": "Server returned an HTTP error",
            "outcome_unknown": "No usable response; server outcome is unknown",
            "invalid_response": "Invalid response structure or JSON; no tools executed",
            "incomplete_response": "Response did not finish normally; no tools executed",
        }
        message = labels.get(category, category)
        if status is not None:
            message += f"（HTTP {status}）"
        super().__init__(message)

    @property
    def retryable(self):
        return self.category == "http_error" and self.status in {429, 502, 503, 504}

def run_with_retry(operation, *, max_retries=2, base_delay=0.5, max_delay=10.0,
                   signal=None, wait=None, random_value=None, on_retry=None):
    if isinstance(max_retries, bool) or not isinstance(max_retries, int) or not 0 <= max_retries <= 10:
        raise ValueError("max_retries must be an integer from 0 to 10")
    if not all(math.isfinite(x) and x >= 0 for x in (base_delay, max_delay)):
        raise ValueError("Retry delays must be finite and nonnegative")
    wait = wait or wait_for_retry
    random_value = random_value or random.random

    for attempt in range(max_retries + 1):
        if signal is not None:
            signal.check()
        try:
            result = operation()
        except RequestFailure as error:
            if not error.retryable or attempt == max_retries:
                raise
            delay = min(max_delay, base_delay * 2 ** attempt)
            delay *= 0.5 + 0.5 * random_value()
            if error.retry_after is not None:
                if error.retry_after > max_delay:
                    raise
                delay = max(delay, error.retry_after)
            if on_retry is not None:
                on_retry(attempt + 1, error.status, delay)
            wait(delay, signal)
        else:
            if signal is not None:
                signal.check()
            return result

def post_json(request, *, signal=None, opener=None):
    opener = opener or urllib.request.urlopen

    def once():
        try:
            with opener(request, timeout=CONFIG["request_timeout_s"]) as response:
                payload = response.read()
        except urllib.error.HTTPError as error:
            try:
                retry_after = parse_retry_after(error.headers.get("Retry-After") if error.headers else None)
                failure = RequestFailure("http_error", status=error.code, retry_after=retry_after)
            finally:
                error.close()
            raise failure from None
        except (urllib.error.URLError, TimeoutError, ConnectionError, http.client.HTTPException):
            raise RequestFailure("outcome_unknown") from None
        try:
            return validate_response(json.loads(payload.decode("utf-8")))
        except (UnicodeError, json.JSONDecodeError):
            raise RequestFailure("invalid_response") from None

    def report(number, status, delay):
        logger.warning("Model request retry %s/%s; HTTP %s; waiting %.2fs", number,
                       CONFIG["request_max_retries"], status, delay)

    return run_with_retry(
        once, max_retries=CONFIG["request_max_retries"],
        base_delay=CONFIG["request_retry_base_s"], max_delay=CONFIG["request_retry_cap_s"],
        signal=signal, on_retry=report,
    )

def validate_response(data):
    try:
        choice = data["choices"][0]
        message = choice["message"]
        if not isinstance(message, dict) or message.get("role") != "assistant":
            raise ValueError("assistant message required")
        if choice.get("finish_reason") not in {"stop", "tool_calls"}:
            raise RequestFailure("incomplete_response")
        calls = message.get("tool_calls") or []
        if not isinstance(calls, list):
            raise ValueError("tool_calls must be a list")
        if (choice['finish_reason'] == 'tool_calls') != bool(calls):
            raise ValueError('finish_reason does not match tool calls')
        identifiers = set()
        for call in calls:
            if not isinstance(call["id"], str) or not call["id"] or call["id"] in identifiers:
                raise ValueError("tool call IDs must be nonempty and unique")
            identifiers.add(call["id"])
            function = call["function"]
            if not isinstance(function["name"], str) or not function["name"]:
                raise ValueError("tool name required")
            arguments = function["arguments"]
            if not isinstance(arguments, str) or not isinstance(json.loads(arguments), dict):
                raise ValueError("tool arguments must encode an object")
        if message.get("content") is not None and not isinstance(message["content"], str):
            raise ValueError("text content must be a string or null")
        if not calls and message.get("content") is None:
            raise ValueError("response contains neither text nor tool calls")
    except (KeyError, IndexError, TypeError, ValueError):
        raise RequestFailure("invalid_response") from None
    return data

def call_llm(messages, tools, system=SYSTEM_PROMPT, signal=None):
    if signal is not None:
        signal.check()

    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(
        CONFIG["base_url"].rstrip("/") + "/chat/completions",
        data=payload,
        method="POST",
        headers={
            "Content-Type": "application/json",
            "Authorization": f"Bearer {CONFIG['api_key']}",
        },
    )

    data = post_json(req, signal=signal)

    if signal is not None:
        signal.check()

    return data

    def test_full_agent_does_not_replay_tool(self):
        tool_reply = llm.fake_tool_response(("write", {"path": "x", "content": "y"}))
        script = [503, tool_reply, 429, llm.fake_text_response("done")]
        bodies, persisted = [], []
        def opener(request, **kwargs):
            bodies.append(request.data)
            item = script.pop(0)
            if isinstance(item, int):
                raise http_error(item)
            return response(item)
        history = [{"role": "user", "content": "test"}]
        with patch.object(transport.urllib.request, "urlopen", side_effect=opener):
            with patch.object(agent.toolbox, "submit", return_value="written") as submit:
                agent.agent_loop(history, emit=lambda *a, **kw: None, on_append=persisted.append)
        self.assertEqual(submit.call_count, 1)
        self.assertEqual(len(history), 4)
        self.assertEqual(len(persisted), 3)
        self.assertEqual(len(bodies), 4)
        self.assertEqual(bodies[0], bodies[1])
        self.assertEqual(bodies[2], bodies[3])
        self.assertEqual(history[2]["tool_call_id"], history[1]["tool_calls"][0]["id"])

    def append(self, kind, **fields):
        record = json.dumps({'kind': kind, **fields}, ensure_ascii=False, allow_nan=False)
        with self.path.open('a', encoding='utf-8', newline='\n') as out:
            out.write(record + '\n')
            out.flush()
            os.fsync(out.fileno())

    def _tool(self, call, *, signal, emit):
        key, fn = call['id'], call['function']
        inp = json.loads(fn['arguments'])
        signal.check()
        self.journal.append('started', id=key)

        emit('tool_start', name=fn['name'], input=inp)
        try:
            output = self.tools.submit(fn['name'], inp, signal=signal)
        except Exception as error:
            output = f'Error: {type(error).__name__}'
        if not isinstance(output, str):
            raise TypeError('Tool must return a string; outcome remains unknown')
        self.journal.append('result', id=key, output=output)
        emit('tool_end', name=fn['name'], output=output)
        return {'tool_call_id': key, 'content': output}

    def restore(self, *, acknowledge_unknown=False):
        events = self.read()
        messages, pending, starts, results, used = [], {}, set(), {}, set()
        for event in events[1:]:
            kind = event['kind']
            if kind == 'message':
                msg = event['message']
                role = msg.get('role')
                if role == 'assistant':
                    if pending:
                        raise RecoveryRequired('Previous tool results are incomplete')
                    for call in msg.get('tool_calls') or []:
                        key = call['id']
                        if key in used:
                            raise RecoveryRequired('Duplicate tool ID within session')
                        used.add(key)
                        pending[key] = call
                elif role == 'tool':
                    key = msg['tool_call_id']
                    if key not in pending:
                        raise RecoveryRequired('Tool result has no matching intent')
                    del pending[key]
                elif role != 'user' or pending:
                    raise RecoveryRequired('Invalid message order')
                messages.append(msg)
            elif kind == 'started':
                key = event['id']
                if key not in pending or key in starts:
                    raise RecoveryRequired('Tool start has no matching intent or is duplicated')
                starts.add(key)
            elif kind == 'result':
                key = event['id']
                if key not in starts or key in results:
                    raise RecoveryRequired('Tool result has no matching start or is duplicated')
                results[key] = event['output']
            else:
                raise RecoveryRequired('Duplicate journal header')
        unknown = [key for key in pending if key in starts and key not in results]
        if unknown and not acknowledge_unknown:
            raise RecoveryRequired('Tool outcome unknown: ' + ', '.join(unknown) +
                                   '; inspect the workspace, then acknowledge explicitly. No automatic replay.')
        for key in pending:
            output = results.get(key, 'Error: outcome unknown; inspect workspace; do not replay'
                                 if key in starts else 'Error: not executed before interruption')
            msg = {'role': 'tool', 'tool_call_id': key, 'content': output}
            self.append('message', message=msg)
            messages.append(msg)
        return messages

    def test_real_crash_unknown_blocks_then_explicit_ack(self):
        result = self.crash('before_result')
        self.assertEqual(result.returncode, 71, result.stderr)
        self.assertEqual((self.root / 'artifact.txt').read_text(), 'once')
        runtime = Runtime(self.journal, provider=lambda *a, **kw: fake_text_response('inspect'))
        before = self.journal.path.read_bytes()
        with self.assertRaises(RecoveryRequired):
            runtime.run()
        self.assertEqual(self.journal.path.read_bytes(), before)
        messages = runtime.run(acknowledge_unknown=True)
        self.assertIn('outcome unknown', messages[-2]['content'])
        self.assertEqual(sum(e['kind'] == 'started' for e in self.journal.read()), 1)

def call_stream(messages, tools, *, system=None, signal, on_delta=None, opener=None):
    body = {'model': CONFIG['model'], 'max_tokens': CONFIG['max_tokens'], 'stream': True,
            'messages': [{'role': 'system', 'content': system or SYSTEM_PROMPT}] + messages}
    if tools:
        body.update(tools=tools, tool_choice='auto')
    request = urllib.request.Request(CONFIG['base_url'].rstrip('/') + '/chat/completions',
        data=json.dumps(body).encode('utf-8'),
        headers={'Content-Type': 'application/json', 'Authorization': 'Bearer ' + CONFIG['api_key']})

    try:
        signal.check()
        with (opener or urllib.request.urlopen)(request, timeout=CONFIG['request_timeout_s']) as response:
            if 'text/event-stream' not in response.headers.get('Content-Type', ''):
                raise RequestFailure('invalid_stream')
            lines = iter(lambda: response.readline(262145), b'')
            return collect_stream(lines, signal=signal, on_delta=on_delta)
    except urllib.error.HTTPError as error:
        status = error.code
        error.close()
        raise RequestFailure('http_error', status=status) from None
    except (urllib.error.URLError, TimeoutError, ConnectionError, http.client.HTTPException):
        raise RequestFailure('outcome_unknown') from None

    def consume(payload):
        nonlocal finish, done
        if payload == '[DONE]':
            done = True
            return
        data = json.loads(payload)
        choices = data['choices']
        if not choices:
            return
        if len(choices) != 1 or choices[0].get('index') != 0 or finish is not None:
            raise ValueError('Expected one choice and no deltas after finish')
        choice = choices[0]
        delta = choice.get('delta', {})
        if delta.get('role', 'assistant') != 'assistant':
            raise ValueError('Invalid role')
        text = delta.get('content')
        if text is not None:
            if not isinstance(text, str):
                raise ValueError('Text delta must be a string')
            content.append(text)
            if on_delta:
                on_delta(text)
        for fragment in delta.get('tool_calls') or []:
            index = fragment['index']
            if type(index) is not int or not 0 <= index < 32:
                raise ValueError('Tool index out of range')
            call = calls.setdefault(index, {'id': '', 'type': 'function',
                                            'function': {'name': '', 'arguments': ''}})
            if fragment.get('type', 'function') != 'function':
                raise ValueError('Unsupported tool type')
            for field in ('id',):
                value = fragment.get(field, '')
                if not isinstance(value, str):
                    raise ValueError('Invalid ID delta')
                call[field] += value
            for field in ('name', 'arguments'):
                value = fragment.get('function', {}).get(field, '')
                if not isinstance(value, str):
                    raise ValueError('Invalid argument delta')
                call['function'][field] += value
        reason = choice.get('finish_reason')
        if reason is not None:
            if reason not in {'stop', 'tool_calls'}:
                raise RequestFailure('incomplete_response')
            finish = reason

def collect_stream(lines, *, signal, on_delta=None):
    content, calls, finish, done, size = [], {}, None, False, 0
    event = []

    def consume(payload):
        nonlocal finish, done
        if payload == '[DONE]':
            done = True
            return
        data = json.loads(payload)
        choices = data['choices']
        if not choices:
            return
        if len(choices) != 1 or choices[0].get('index') != 0 or finish is not None:
            raise ValueError('Expected one choice and no deltas after finish')
        choice = choices[0]
        delta = choice.get('delta', {})
        if delta.get('role', 'assistant') != 'assistant':
            raise ValueError('Invalid role')
        text = delta.get('content')
        if text is not None:
            if not isinstance(text, str):
                raise ValueError('Text delta must be a string')
            content.append(text)
            if on_delta:
                on_delta(text)
        for fragment in delta.get('tool_calls') or []:
            index = fragment['index']
            if type(index) is not int or not 0 <= index < 32:
                raise ValueError('Tool index out of range')
            call = calls.setdefault(index, {'id': '', 'type': 'function',
                                            'function': {'name': '', 'arguments': ''}})
            if fragment.get('type', 'function') != 'function':
                raise ValueError('Unsupported tool type')
            for field in ('id',):
                value = fragment.get(field, '')
                if not isinstance(value, str):
                    raise ValueError('Invalid ID delta')
                call[field] += value
            for field in ('name', 'arguments'):
                value = fragment.get('function', {}).get(field, '')
                if not isinstance(value, str):
                    raise ValueError('Invalid argument delta')
                call['function'][field] += value
        reason = choice.get('finish_reason')
        if reason is not None:
            if reason not in {'stop', 'tool_calls'}:
                raise RequestFailure('incomplete_response')
            finish = reason

    try:
        for raw in lines:
            signal.check()
            size += len(raw)
            if size > 2_000_000 or len(raw) > 262144:
                raise ValueError('Stream exceeds size limit')
            line = raw.decode('utf-8').rstrip('\r\n')
            if not line:
                if event:
                    consume('\n'.join(event))
                    event = []
                    if done:
                        break
            elif line.startswith('data:'):
                event.append(line[5:].lstrip(' '))
        signal.check()
        if not done or finish is None or sorted(calls) != list(range(len(calls))):
            raise RequestFailure('incomplete_stream')
        if (finish == 'tool_calls') != bool(calls):
            raise ValueError('finish_reason does not match tool calls')
        message = {'role': 'assistant', 'content': ''.join(content) if content else None}
        if calls:
            message['tool_calls'] = [calls[i] for i in sorted(calls)]
        return validate_response({'choices': [{'message': message, 'finish_reason': finish}]})
    except (ValueError, KeyError, TypeError, AttributeError, UnicodeError):
        raise RequestFailure('invalid_stream') from None

    def _request(self, messages, tools, *, system, signal):
        if self.provider:
            response = self.provider(messages, tools, system=system, signal=signal)
        elif self.stream:
            response = call_stream(messages, tools, system=system, signal=signal,
                                   on_delta=lambda text: self.emit('text_delta', text=text))
        else:
            response = llm.call_llm(messages, tools, system=system, signal=signal)
        signal.check()
        response = validate_response(response)
        used = {c['id'] for m in messages for c in m.get('tool_calls') or []}
        if any(c['id'] in used for c in llm.extract_tool_calls(response)):
            raise RecoveryRequired('Provider reused a tool ID; execution refused')
        return response

    def __init__(self, root, *, allow_write=False):
        self.root = Path(root).resolve(strict=True)
        if not self.root.is_dir():
            raise ValueError('Workspace root must be a directory')
        self.allow_write = allow_write
        self.reserved = set()
        self.registry = {}
        for schema in TOOLS:
            name = schema['function']['name']
            if name in {'read', 'write', 'edit'}:
                self.register(name, schema, getattr(self, '_' + name), mutating=name != 'read')

    def submit(self, name, inp, *, signal):
        signal.check()
        if name not in self.registry:
            raise PermissionError('Tool is not registered')
        _, handler, write = self.registry[name]
        if write and not self.allow_write:
            raise PermissionError('Write access requires --allow-write')
        if not isinstance(inp, dict):
            raise ValueError('Tool arguments must be an object')
        return handler(inp, signal=signal)

    def path(self, value):
        if not isinstance(value, str) or not value:
            raise ValueError('path must be a nonempty string')
        path = (self.root / value).resolve()
        if not path.is_relative_to(self.root):
            raise PermissionError('Path is outside the workspace')
        relative = path.relative_to(self.root)
        if any(':' in part or part.endswith((' ', '.')) or
               part.split('.')[0].upper() in {'CON', 'PRN', 'AUX', 'NUL',
                   *(f'COM{i}' for i in range(1, 10)), *(f'LPT{i}' for i in range(1, 10))}
               for part in relative.parts):
            raise PermissionError('Device paths, alternate data streams, and ambiguous names are not supported')
        if (relative.parts and relative.parts[0].lower() in {'.mini_agent', '.git'}) or path in self.reserved:
            raise PermissionError('Model tools cannot access journals or Git metadata')
        return path

    def _write(self, inp, *, signal):
        path, content = self.path(inp.get('path')), inp.get('content')
        if not isinstance(content, str):
            raise ValueError('content must be a string')
        signal.check()
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(content, encoding='utf-8')
        return f'Wrote {len(content)} characters'

                def provider(messages, tools, **kwargs):
                    nonlocal requests
                    requests += 1
                    if requests == 1:
                        if task == 'read':
                            return fake_tool_response(('read', {'path': 'input.txt', 'offset': 2, 'limit': 1}))
                        return fake_tool_response(('write', {'path': 'answer.txt', 'content': 'hello'}))
                    if task == 'write_edit' and requests == 2:
                        return fake_tool_response(('edit', {'path': 'answer.txt',
                                                           'old_string': 'hello', 'new_string': 'hello world'}))
                    return fake_text_response(messages[-1]['content'])

def evaluate(repeats=3):
    if type(repeats) is not int or repeats < 1:
        raise ValueError('repeats must be a positive integer')
    rows = []
    for task in ('read', 'write_edit', 'deny_write'):
        for repeat in range(1, repeats + 1):
            with tempfile.TemporaryDirectory(prefix='mini-agent-eval-') as tmp:
                root = Path(tmp)
                (root / 'input.txt').write_text('alpha\nbeta\n', encoding='utf-8')
                journal = Journal.create(root / '.mini_agent' / 'run.jsonl', root)
                requests, tool_count = 0, 0
                def provider(messages, tools, **kwargs):
                    nonlocal requests
                    requests += 1
                    if requests == 1:
                        if task == 'read':
                            return fake_tool_response(('read', {'path': 'input.txt', 'offset': 2, 'limit': 1}))
                        return fake_tool_response(('write', {'path': 'answer.txt', 'content': 'hello'}))
                    if task == 'write_edit' and requests == 2:
                        return fake_tool_response(('edit', {'path': 'answer.txt',
                                                           'old_string': 'hello', 'new_string': 'hello world'}))
                    return fake_text_response(messages[-1]['content'])
                def emit(kind, **kw):
                    nonlocal tool_count
                    if kind == 'tool_start':
                        tool_count += 1
                start = time.perf_counter()
                messages, error_kind = [], None
                try:
                    messages = Runtime(journal, provider=provider, allow_write=task == 'write_edit',
                                       emit=emit).run('Execute the task contract', max_turns=4)
                except Exception as error:
                    error_kind = type(error).__name__
                elapsed = time.perf_counter() - start
                answer = root / 'answer.txt'

                passed = not error_kind and bool(messages) and ((task == 'read' and messages[-1]['content'] == '2: beta\n') or
                          (task == 'write_edit' and answer.is_file() and answer.read_text(encoding='utf-8') == 'hello world') or
                          (task == 'deny_write' and not answer.exists() and
                           'PermissionError' in messages[-1]['content']))
                rows.append({'task': task, 'repeat': repeat, 'passed': passed,
                             'error': error_kind,
                             'requests': requests, 'tools': tool_count, 'elapsed_s': elapsed,
                             'tokens': None, 'cost': None})
    return {'mode': 'scripted-control-flow', 'real_model': False,
            'attempts': rows, 'summary': {'passed': sum(r['passed'] for r in rows),
            'total': len(rows), 'pass_rate': sum(r['passed'] for r in rows) / len(rows),
            'mean_elapsed_s': statistics.mean(r['elapsed_s'] for r in rows)},
            'limitations': ['Scripted providers do not measure model accuracy', 'Real tokens and costs were not measured',
                            'This small task set is not a general coding benchmark']}

def main():
    parser = argparse.ArgumentParser(description='Run a deterministic offline control-flow evaluation')
    parser.add_argument('--repeats', type=int, default=3)
    parser.add_argument('--output', type=Path)
    args = parser.parse_args()
    report = evaluate(args.repeats)
    value = json.dumps(report, ensure_ascii=False, indent=2)
    if args.output:
        args.output.parent.mkdir(parents=True, exist_ok=True)
        args.output.write_text(value + '\n', encoding='utf-8')
    print(value)
    return 0 if report['summary']['passed'] == report['summary']['total'] else 1

    def test_evaluation_records_runtime_failures(self):
        from .evaluation import evaluate
        with patch('mini_agent.evaluation.Runtime.run', side_effect=RuntimeError('private-detail')):
            report = evaluate(1)
        self.assertEqual(report['summary']['passed'], 0)
        self.assertEqual(report['summary']['total'], 3)
        self.assertTrue(all(r['error'] == 'RuntimeError' for r in report['attempts']))
        self.assertNotIn('private-detail', json.dumps(report))

    def __init__(self, journal, *, provider=None, allow_write=False, stream=False, emit=None):
        self.journal = journal
        header = journal.read()[0]
        self.tools = WorkspaceTools(header['root'], allow_write=allow_write)
        self.tools.reserved.update({journal.path, journal.path.with_name(journal.path.name + '.lock')})

        self.signal = Signal()
        self.provider = provider
        self.stream = stream
        self.emit = emit or (lambda kind, **kw: None)

    def run(self, text=None, *, max_turns=8, acknowledge_unknown=False):
        if type(max_turns) is not int or max_turns < 1:
            raise ValueError('max_turns must be a positive integer')
        with self.journal.locked():
            self.signal.check()
            messages = self.journal.restore(acknowledge_unknown=acknowledge_unknown)
            if text is not None:
                if not isinstance(text, str) or not text.strip():
                    raise ValueError('Task must not be empty')
                msg = {'role': 'user', 'content': text}
                self.journal.append('message', message=msg)
                messages.append(msg)
            if not messages:
                raise ValueError('A new session requires a task')

            if text is None and messages[-1]['role'] == 'assistant' and not messages[-1].get('tool_calls'):
                return messages
            return agent.agent_loop(messages, tools=self.tools.definitions(), signal=self.signal,
                system='You are a coding agent. Use only the provided tools. '
                       'Tool outcomes marked unknown require human inspection; never replay them.',
                max_turns=max_turns, emit=self.emit, request_fn=self._request, tool_runner=self._tool,
                on_append=lambda msg: self.journal.append('message', message=msg))

    def register(self, name, schema, handler, *, mutating=True):
        if name in self.registry or schema['function']['name'] != name or not callable(handler):
            raise ValueError('Duplicate tool or invalid registration')
        self.registry[name] = (deepcopy(schema), handler, mutating)

def main(argv=None):
    setup_console()
    parser = argparse.ArgumentParser(description='mini agent-2: read-only by default, no shell tool')
    parser.add_argument('prompt', nargs='?')
    parser.add_argument('--root', default='.', help='Root for new sessions; resume uses the journal root')
    parser.add_argument('--resume', type=Path, help='Resume a journal (not a v1 Session JSONL file)')
    parser.add_argument('--status', action='store_true', help='Print execution stages without model requests')
    parser.add_argument('--stream', action='store_true')
    parser.add_argument('--allow-write', action='store_true', help='Allow write/edit for this run')
    parser.add_argument('--acknowledge-unknown', action='store_true', help='Acknowledge inspected unknown outcomes and continue')
    parser.add_argument('--max-turns', type=int, default=8)
    args = parser.parse_args(argv)
    if args.status:
        if not args.resume:
            parser.error('--status requires --resume')
        journal = Journal(args.resume)
        with journal.locked():
            for event in journal.read():
                if event['kind'] in {'started', 'result'}:
                    print(event['kind'], event['id'])
        return 0
    if args.resume is None and not args.prompt:
        parser.error('A new session requires task text')
    if args.max_turns < 1:
        parser.error('--max-turns must be positive')
    require_api_key()
    journal = Journal(args.resume) if args.resume else Journal.create(
        Path(args.root) / '.mini_agent' / 'advanced' / (uuid.uuid4().hex + '.jsonl'), args.root)
    print('Journal: ', journal.path)
    print('Streamed text is a preview; only complete validated responses enter history.' if args.stream else 'Non-streaming mode')
    finished = False
    def render(kind, **kw):
        nonlocal finished
        if kind == 'text_delta':
            print(kw['text'], end='', flush=True)
        elif kind == 'text' and not args.stream:
            print(kw['text'])
        elif kind == 'done':
            finished = True
        elif kind in {'aborted', 'max_turns'}:
            print('\nRun stopped: ', kind)
    runtime = Runtime(journal, allow_write=args.allow_write, stream=args.stream, emit=render)
    try:
        messages = runtime.run(args.prompt, max_turns=args.max_turns,
                              acknowledge_unknown=args.acknowledge_unknown)
        if messages[-1]['role'] == 'assistant' and not messages[-1].get('tool_calls'):
            finished = True
        return 0 if finished else 2
    except KeyboardInterrupt:
        runtime.signal.abort('Ctrl+C')
        print('\nStopped; inspect the journal before resuming. Preview text is not committed.')
        return 130
    except (RecoveryRequired, RuntimeError, OSError, ValueError) as error:
        print('\nRun incomplete: ', str(error) if isinstance(error, (RecoveryRequired, RequestFailure))
              else type(error).__name__)
        return 1