feat(agent): gate completion on observable verification conditions

assets/tools_schema.json

@@ -55,7 +55,10 @@
     "name": "update_working_checkpoint",
     "description": "Short-term working notepad (auto-injected each turn). Call early/mid task only when there is non-obvious state worth keeping. When: (1) after reading SOP — store needs & constraints (skip tasks <5 steps); (2) before subtask switch / context flush; (3) after repeated failures — re-read SOP and store new findings; (4) on new task — replace notepad, drop old progress, keep still-valid constraints.\n\nDon't call: tasks <5 steps, greetings, or when the task is done (use start_long_term_update instead).",
     "parameters": {"type": "object", "properties": {
-      "key_info": {"type": "string", "description": "Replaces current notepad (<200 tokens). Incremental update: review existing, keep valid, add/remove/modify. Store: pitfalls, user requirements, key params/findings, file paths, progress, next steps. Don't store: ephemeral info, obvious context, old task info when user switched tasks. Prefer over-updating over losing key info"}}}
+      "key_info": {"type": "string", "description": "Replaces current notepad (<200 tokens). Incremental update: review existing, keep valid, add/remove/modify. Store: pitfalls, user requirements, key params/findings, file paths, progress, next steps. Don't store: ephemeral info, obvious context, old task info when user switched tasks. Prefer over-updating over losing key info"},
+      "verify_conditions": {"type": "array", "description": "Observable completion conditions to define before execution. Any unmet condition blocks completion. Kinds: file_exists(path), file_contains(path, substring), last_exit_code_zero, tool_result_contains(tool, substring).", "items": {"type": "object", "required": ["kind"], "properties": {
+        "kind": {"type": "string", "enum": ["file_exists", "file_contains", "last_exit_code_zero", "tool_result_contains"]},
+        "description": {"type": "string"}, "path": {"type": "string"}, "tool": {"type": "string"}, "substring": {"type": "string"}}}}}}
   }},
   {"type": "function", "function": {
     "name": "ask_user",

assets/tools_schema_cn.json

@@ -56,7 +56,10 @@
     "description": "短期工作便签，每轮自动注入上下文，防长任务信息丢失。前中期调用，非结束时。何时调用：(1)任务开始读SOP后，存用户需求和关键约束/参数（简单1-2步任务除外）；(2)子任务切换或上下文即将被冲刷前；(3)多次重试失败后，重读SOP并必须调用存储新发现；(4)切换新任务时更新内容，清旧进度但保留仍有效的约束。\n\n何时不调用：简单任务（1-2步且无严重约束）、任务已完成时（应当用长期结算工具）",
     "parameters": {"type": "object", "properties": {
       "key_info": {"type": "string", "description": "替换当前便签（<200 tokens）。增量更新：先回顾现有内容，保留仍有效的，再增删改。存：要避的坑、用户原始需求、关键参数/发现、文件路径、当前进度、下一步计划。不存：马上要用用完即丢的、上下文中显而易见的、用户已换全新任务时的旧任务信息。宁多更新不丢关键"},
-      "related_sop": {"type": "string", "description": "相关sop名称，可以多个，必要时需要再读"}}}
+      "related_sop": {"type": "string", "description": "相关sop名称，可以多个，必要时需要再读"},
+      "verify_conditions": {"type": "array", "description": "执行前定义的可观察完成条件；任一条件未满足都会阻止任务结束。kind 支持 file_exists(path)、file_contains(path, substring)、last_exit_code_zero、tool_result_contains(tool, substring)。", "items": {"type": "object", "required": ["kind"], "properties": {
+        "kind": {"type": "string", "enum": ["file_exists", "file_contains", "last_exit_code_zero", "tool_result_contains"]},
+        "description": {"type": "string"}, "path": {"type": "string"}, "tool": {"type": "string"}, "substring": {"type": "string"}}}}}}
   }},
   {"type": "function", "function": {
     "name": "ask_user",
@@ -70,4 +73,4 @@
     "description": "准备开始提炼记忆。发现值得长期记忆的信息（环境事实/用户偏好/避坑经验）时调用此工具。已记忆更新或在自主流程内时无需调用。超15轮完成的任务必须调用以沉淀经验",
     "parameters": {"type": "object", "properties": {}}}
   }
-]
+]

ga.py

@@ -277,6 +277,25 @@ def consume_file(dr, file):
         os.remove(os.path.join(dr, file))
         return content
 
+def _verify_condition(condition, evidence, resolve_path=os.path.abspath):
+    kind = condition.get('kind'); raw_path = condition.get('path', '')
+    path = resolve_path(raw_path) if raw_path else ''
+    if kind == 'file_exists': return bool(path) and os.path.exists(path)
+    if kind == 'file_contains':
+        needle = condition.get('substring', '')
+        if not needle or not os.path.isfile(path): return False
+        try:
+            with open(path, encoding='utf-8', errors='replace') as f: return needle in f.read()
+        except OSError: return False
+    if kind == 'last_exit_code_zero':
+        try: result = json.loads(evidence.get('code_run', ''))
+        except (json.JSONDecodeError, TypeError): return False
+        return isinstance(result, dict) and result.get('exit_code') == 0
+    if kind == 'tool_result_contains':
+        tool = condition.get('tool'); needle = condition.get('substring', '')
+        return bool(tool and needle) and needle in str(evidence.get(tool, ''))
+    return False
+
 class GenericAgentHandler(BaseHandler):
     '''Generic Agent 工具库，包含多种工具的实现。工具函数自动加上了 do_ 前缀。实际工具名没有前缀。'''
     def __init__(self, parent, last_history=None, cwd='./temp'):
@@ -462,6 +481,10 @@ def do_update_working_checkpoint(self, args, response):
         '''为整个任务设定后续需要临时记忆的重点。'''
         key_info = args.get("key_info", "")
         if "key_info" in args: self.working['key_info'] = key_info
+        if "verify_conditions" in args:
+            conditions = args['verify_conditions'] or []
+            valid = isinstance(conditions, list) and all(isinstance(c, dict) for c in conditions)
+            self.working['verify_conditions'] = conditions if valid else [{'kind': 'invalid'}]
         self.working['passed_sessions'] = 0
         yield f"[Info] Updated key_info.\n"
         next_prompt = self._get_anchor_prompt(skip=args.get('_index', 0) > 0)
@@ -474,6 +497,11 @@ def _retry_or_exit(self, prompt):
         if self._empty_ct >= 3: return StepOutcome({}, should_exit=True)
         return StepOutcome({}, next_prompt=prompt)
 
+    def _unsatisfied_verify_conditions(self):
+        evidence = self.working.get('_tool_evidence', {})
+        return [c for c in self.working.get('verify_conditions', [])
+                if not _verify_condition(c, evidence, self._get_abs_path)]
+
     def do_no_tool(self, args, response):
         '''这是一个特殊工具，由引擎自主调用，不要包含在TOOLS_SCHEMA里。
         当模型在一轮中未显式调用任何工具时，由引擎自动触发。
@@ -515,7 +543,12 @@ def do_no_tool(self, args, response):
                         "并明确是否还需要额外的实际操作。"
                     )
                     return StepOutcome({}, next_prompt=next_prompt)
-
+
+        if pending := self._unsatisfied_verify_conditions():
+            yield f"[Warn] {len(pending)} verify condition(s) are not satisfied.\n"
+            details = '\n'.join(f"- {c.get('description') or c.get('kind')}" for c in pending)
+            return StepOutcome({}, next_prompt="[VERIFY] Completion blocked by unmet conditions:\n" + details)
+
         if self._in_plan_mode():
             remaining = self._check_plan_completion()
             if remaining == 0:
@@ -577,6 +610,10 @@ def turn_end_callback(self, response, tool_calls, tool_results, turn, next_promp
             self.history_info.append('[Agent] ' + summary)
         if not rsumm and tool_calls and tool_calls[0]['tool_name'] != 'no_tool':
             next_prompt += "\n\n\n[TIPS] 必须在回复文本中包含<summary>！\n\n"
+        by_id = {r['tool_use_id']: r['content'] for r in tool_results}
+        evidence = self.working.setdefault('_tool_evidence', {})
+        for call in tool_calls:
+            if call.get('id') in by_id: evidence[call['tool_name']] = by_id[call['id']]
         _plan = self._in_plan_mode()
 
         if turn % 175 == 0 and (not _plan):

tests/test_verify_conditions.py

@@ -0,0 +1,79 @@
+import json
+from pathlib import Path
+from types import SimpleNamespace
+
+from ga import GenericAgentHandler, _verify_condition
+
+
+def response(content='done'):
+    return SimpleNamespace(content=content, thinking='')
+
+
+def exhaust(generator):
+    try:
+        while True: next(generator)
+    except StopIteration as result: return result.value
+
+
+def handler(tmp_path):
+    parent = SimpleNamespace(extrakeyinfo=None, intervene=None, task_dir=None, _turn_end_hooks={})
+    return GenericAgentHandler(parent, cwd=str(tmp_path))
+
+
+def test_file_conditions_use_handler_working_directory(tmp_path):
+    target = tmp_path / 'report.txt'
+    target.write_text('verified output')
+    resolve = lambda path: str(tmp_path / path)
+    assert _verify_condition({'kind': 'file_exists', 'path': 'report.txt'}, {}, resolve)
+    assert _verify_condition({'kind': 'file_contains', 'path': 'report.txt',
+                              'substring': 'verified'}, {}, resolve)
+    assert not _verify_condition({'kind': 'file_exists'}, {}, resolve)
+
+
+def test_exit_code_condition_uses_real_code_run_result():
+    passed = {'code_run': json.dumps({'status': 'success', 'exit_code': 0})}
+    failed = {'code_run': json.dumps({'status': 'error', 'exit_code': 1})}
+    condition = {'kind': 'last_exit_code_zero'}
+    assert _verify_condition(condition, passed)
+    assert not _verify_condition(condition, failed)
+
+
+def test_tool_result_condition_requires_matching_tool_output():
+    condition = {'kind': 'tool_result_contains', 'tool': 'file_read', 'substring': '42 rows'}
+    assert _verify_condition(condition, {'file_read': 'exported 42 rows'})
+    assert not _verify_condition(condition, {'code_run': 'exported 42 rows'})
+
+
+def test_unmet_condition_blocks_no_tool_completion(tmp_path):
+    agent = handler(tmp_path)
+    agent.working['verify_conditions'] = [
+        {'kind': 'file_exists', 'path': 'missing.txt', 'description': 'report exists'}]
+    for _ in range(6):
+        outcome = exhaust(agent.do_no_tool({}, response()))
+        assert outcome.next_prompt == '[VERIFY] Completion blocked by unmet conditions:\n- report exists'
+
+
+def test_tool_result_is_recorded_and_allows_completion(tmp_path):
+    agent = handler(tmp_path)
+    agent.working['verify_conditions'] = [{'kind': 'last_exit_code_zero'}]
+    agent.turn_end_callback(response('<summary>tests ran</summary>'),
+                            [{'tool_name': 'code_run', 'id': 'call-1'}],
+                            [{'tool_use_id': 'call-1', 'content': json.dumps({'exit_code': 0})}],
+                            1, '', {})
+    outcome = exhaust(agent.do_no_tool({}, response()))
+    assert outcome.next_prompt is None
+
+
+def test_checkpoint_registers_conditions(tmp_path):
+    agent = handler(tmp_path); agent.current_turn = 2
+    conditions = [{'kind': 'file_exists', 'path': 'report.txt'}]
+    exhaust(agent.do_update_working_checkpoint({'verify_conditions': conditions}, response()))
+    assert agent.working['verify_conditions'] == conditions
+
+
+def test_conditions_are_exposed_in_both_tool_schemas():
+    for path in ('assets/tools_schema.json', 'assets/tools_schema_cn.json'):
+        schema = json.loads(Path(path).read_text())
+        checkpoint = next(tool['function'] for tool in schema
+                          if tool['function']['name'] == 'update_working_checkpoint')
+        assert 'verify_conditions' in checkpoint['parameters']['properties']