附录 E：完整测试与实跑记录

from pathlib import Path
import json
import sys
import tempfile
import threading
import unittest
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from harness import (BudgetExceeded, Call, DemoModel, Engine, HarnessError, Journal, Limits,
                     Policy, ProtocolError, Reply, ReviewRequired, Tools, TransientModelError,
                     VerificationFailed, decode_stream, digest)

class Replies:
    def __init__(self, replies):
        self.replies = replies
        self.requests = 0
    def stream(self, messages, step):
        self.requests += 1
        reply = self.replies[step]
        yield {'type': 'text.delta', 'text': reply.text}
        for call in reply.calls:
            yield {'type': 'tool.delta', 'id': call.id, 'name': call.name,
                   'arguments': json.dumps(call.arguments)}
        yield {'type': 'reply.end', 'reason': 'stop'}

class HarnessTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.addCleanup(self.temp.cleanup)
        self.root = Path(self.temp.name)
        self.file = self.root / 'hello.txt'
        self.file.write_text('helo, harness\n')
        self.journal = Journal(self.root / '.harness' / 'events.jsonl')
        self.tools = Tools(self.root, Policy(allow_write=True))
    def valid(self):
        return self.file.read_text() == 'hello, harness\n'
    def engine(self, model=None, **kwargs):
        return Engine(self.journal, self.tools, model or DemoModel(), **kwargs)
    def test_demo_has_real_effect_and_verified_end(self):
        self.engine().run('fix', self.valid)
        self.assertTrue(self.valid())
        self.assertEqual(self.journal.events[-1]['type'], 'run.finished')
        self.assertEqual([e['data']['status'] for e in self.engine().events('tool.result')], ['ok'] * 3)
    def test_resume_finished_session_does_not_repeat_tools(self):
        self.engine().run('fix', self.valid)
        count = len(self.engine().events('tool.started'))
        model = Replies([])
        self.engine(model).run('fix', self.valid)
        self.assertEqual(model.requests, 0)
        self.assertEqual(len(self.engine().events('tool.started')), count)
    def test_saved_completion_is_revalidated(self):
        self.engine().run('fix', self.valid)
        self.file.write_text('changed')
        with self.assertRaises(VerificationFailed):
            self.engine().run('fix', self.valid)
    def test_final_text_is_not_completion_evidence(self):
        with self.assertRaises(VerificationFailed):
            self.engine(Replies([Reply('I fixed it')])).run('fix', self.valid)
        self.assertFalse(self.engine().events('run.finished'))
    def test_read_only_blocks_side_effect(self):
        with self.assertRaises(VerificationFailed):
            Engine(self.journal, Tools(self.root, Policy()), DemoModel()).run('fix', self.valid)
        self.assertEqual(self.file.read_text(), 'helo, harness\n')
        self.assertIn('denied', [e['data']['status'] for e in self.engine().events('tool.result')])
    def test_parent_traversal_denied(self):
        self.assertEqual(self.tools.execute(Call('x', 'read_file', {'path': '../outside'}))['status'], 'denied')
    def test_symlink_escape_denied(self):
        (self.root / 'escape').symlink_to(self.root.parent)
        self.assertEqual(self.tools.execute(Call('x', 'read_file', {'path': 'escape/anything'}))['status'], 'denied')
    def test_reserved_journal_denied(self):
        result = self.tools.execute(Call('x', 'write_file', {'path': '.harness/events.jsonl', 'text': 'oops',
                                                          'expected_sha256': 'missing'}))
        self.assertEqual(result['status'], 'denied')
    def test_bad_schema_and_unknown_tool_never_dispatch(self):
        for call in [Call('a', 'shell', {'command': 'ignored'}), Call('b', 'read_file', {'path': 1}),
                     Call('c', 'read_file', {'path': 'hello.txt', 'extra': True})]:
            self.assertEqual(self.tools.execute(call)['status'], 'invalid')
    def test_stale_hash_prevents_write(self):
        result = self.tools.execute(Call('x', 'write_file', {'path': 'hello.txt', 'text': 'replacement',
                                                          'expected_sha256': 'stale'}))
        self.assertEqual(result['status'], 'conflict')
        self.assertEqual(self.file.read_text(), 'helo, harness\n')
    def test_truncated_model_stream_cannot_execute(self):
        class Partial:
            def stream(self, messages, step):
                yield {'type': 'tool.delta', 'id': 'x', 'name': 'write_file', 'arguments': '{"path":'}
                yield {'type': 'reply.end', 'reason': 'length'}
        with self.assertRaises(ProtocolError):
            self.engine(Partial()).run('fix', self.valid)
        self.assertFalse(self.engine().events('tool.started'))
    def test_missing_stream_end_rejected(self):
        with self.assertRaises(ProtocolError):
            decode_stream([{'type': 'text.delta', 'text': 'partial'}])
    def test_fragmented_json_reassembled(self):
        reply = decode_stream([{'type': 'tool.delta', 'id': 'x', 'name': 'read_file', 'arguments': '{"path":'},
                               {'type': 'tool.delta', 'id': 'x', 'arguments': '"hello.txt"}'},
                               {'type': 'reply.end', 'reason': 'stop'}])
        self.assertEqual(reply.calls[0].arguments, {'path': 'hello.txt'})
    def test_duplicate_call_id_in_later_reply_stops(self):
        call = Call('repeat', 'read_file', {'path': 'hello.txt'})
        model = Replies([Reply('', (call,)), Reply('', (call,))])
        with self.assertRaises(ProtocolError):
            self.engine(model).run('fix', self.valid)
        self.assertEqual(len(self.engine().events('tool.started')), 1)
    def test_model_request_budget_is_enforced(self):
        with self.assertRaises(BudgetExceeded):
            self.engine(limits=Limits(model_requests=1)).run('fix', self.valid)
        self.assertFalse(self.valid())
    def test_tool_budget_stops_before_second_side_effect(self):
        with self.assertRaises(BudgetExceeded):
            self.engine(limits=Limits(tool_calls=1)).run('fix', self.valid)
        self.assertEqual(len(self.engine().events('tool.started')), 1)
        self.assertFalse(self.valid())
    def test_unknown_result_blocks_reexecution(self):
        call = Call('maybe-written', 'write_file', {'path': 'hello.txt', 'text': 'unseen',
                                                  'expected_sha256': digest(self.file.read_bytes())})
        with self.journal.writer():
            self.journal.append('session.started', version=1, workspace=str(self.root.resolve()),
                                policy={'allow_write': True, 'max_read_bytes': 65536})
            self.journal.append('user.input', text='fix')
            self.journal.append('model.reply', text='', calls=[{'id': call.id, 'name': call.name, 'arguments': call.arguments}])
            self.journal.append('tool.started', id=call.id, name=call.name, arguments=call.arguments)
        self.file.write_text('already changed')
        model = Replies([])
        with self.assertRaises(ReviewRequired):
            self.engine(model).run('fix', self.valid)
        self.assertEqual(self.file.read_text(), 'already changed')
        self.assertEqual(model.requests, 0)
        self.assertEqual(self.engine().events('tool.result')[0]['data']['status'], 'unknown')
    def test_preview_preserves_full_artifact(self):
        self.file.write_text('甲' * 3000)
        model = Replies([Reply('', (Call('large', 'read_file', {'path': 'hello.txt'}),)), Reply('done')])
        self.engine(model, limits=Limits(preview_chars=100)).run('inspect', lambda: True)
        result = self.engine().events('tool.result')[0]['data']
        self.assertIn('[truncated]', result['content'])
        raw = (self.journal.path.parent / result['artifact']).read_bytes()
        self.assertEqual(digest(raw), result['sha256'])
        self.assertIn('甲' * 3000, raw.decode())
    def test_cancellation_is_not_success(self):
        cancel = threading.Event();cancel.set()
        with self.assertRaises(HarnessError):
            self.engine(cancel=cancel).run('fix', self.valid)
        self.assertFalse(self.engine().events('tool.started'))
        self.assertFalse(self.engine().events('run.finished'))
    def test_writer_lock_excludes_another_writer(self):
        with self.journal.writer():
            with self.assertRaises(HarnessError):
                with Journal(self.journal.path).writer():
                    pass
    def test_torn_tail_is_not_silently_discarded(self):
        self.journal.path.write_bytes(b'{"incomplete":')
        with self.assertRaises(ProtocolError):
            Journal(self.journal.path)
    def test_tampered_record_is_detected(self):
        with self.journal.writer():
            self.journal.append('example', text='original')
        self.journal.path.write_text(self.journal.path.read_text().replace('original', 'modified'))
        with self.assertRaises(ProtocolError):
            Journal(self.journal.path)
    def test_transient_model_error_retried_without_tool_retry(self):
        class Once(DemoModel):
            attempts = 0
            def stream(self, messages, step):
                self.attempts += 1
                if self.attempts == 1:
                    raise TransientModelError('temporary pre-reply failure')
                yield from super().stream(messages, step)
        self.engine(Once()).run('fix', self.valid)
        self.assertEqual(len(self.engine().events('model.retryable_error')), 1)
        self.assertEqual(len(self.engine().events('tool.started')), 3)
    def test_goal_or_policy_cannot_change_on_resume(self):
        self.engine().run('fix', self.valid)
        with self.assertRaises(ProtocolError):
            self.engine().run('different goal', self.valid)
        with self.assertRaises(ProtocolError):
            Engine(self.journal, Tools(self.root, Policy()), DemoModel()).run('fix', self.valid)

if __name__ == '__main__':
    unittest.main()

{
  "success": {
    "status": "verified",
    "answer": "工具阶段结束，是否完成由宿主验收。",
    "events": 18
  },
  "resume": {
    "status": "verified",
    "answer": "工具阶段结束，是否完成由宿主验收。",
    "events": 18
  },
  "read_only": {
    "status": "stopped",
    "reason": "model answer does not satisfy acceptance"
  },
  "read_only_exit": 2,
  "workspace_scope": "temporary directories; no model API called"
}