Codex CLI 日常工作流

"""Collect fixed, model-free checks; fixtures are data, never commands."""
import argparse
import json
from pathlib import Path
import shutil
import subprocess

CODEX_CHECKS = {
    'version': ['--version'],
    'help': ['--help'],
    'resume': ['resume', '--help'],
    'fork': ['fork', '--help'],
    'review': ['review', '--help'],
}
GIT_CHECKS = {
    'root': ['rev-parse', '--show-toplevel'],
    'branch': ['branch', '--show-current'],
    'status': ['status', '--porcelain=v1', '--untracked-files=normal'],
}
NAMES = set(CODEX_CHECKS) | set(GIT_CHECKS)


def run_fixed(binary, args, cwd):
    if not binary:
        return {'rc': None, 'out': '', 'error': 'missing executable'}
    try:
        r = subprocess.run(
            [binary, *args], cwd=cwd, capture_output=True,
            text=True, encoding='utf-8', errors='strict', timeout=10, check=False,
        )
    except subprocess.TimeoutExpired:
        return {'rc': None, 'out': '', 'error': 'timeout'}
    except UnicodeDecodeError:
        return {'rc': None, 'out': '',
                'error': 'invalid UTF-8 command output; result unavailable'}
    except OSError as e:
        return {'rc': None, 'out': '', 'error': type(e).__name__}
    return {'rc': r.returncode, 'out': r.stdout, 'error': r.stderr}


def collect(repo):
    root = Path(repo).expanduser().resolve(strict=True)
    if not root.is_dir():
        raise ValueError('repo must be a directory')
    codex, git = shutil.which('codex'), shutil.which('git')
    checks = {k: run_fixed(codex, v, root)
              for k, v in CODEX_CHECKS.items()}
    prefix = ['--no-optional-locks', '-c', 'core.fsmonitor=false']
    checks.update({k: run_fixed(git, prefix + v, root)
                   for k, v in GIT_CHECKS.items()})
    return {'schema': 1, 'kind': 'local_probe',
            'cwd': str(root), 'checks': checks}


def validate(data):
    if not isinstance(data, dict) or type(data.get('schema')) is not int:
        raise ValueError('schema must be integer 1')
    if data['schema'] != 1:
        raise ValueError('unsupported schema')
    if data.get('kind') not in ('local_probe', 'synthetic_fixture'):
        raise ValueError('unknown evidence kind')
    if not isinstance(data.get('cwd'), str) or not data['cwd']:
        raise ValueError('cwd is required')
    checks = data.get('checks')
    if not isinstance(checks, dict) or set(checks) != NAMES:
        raise ValueError('unexpected or missing check name')
    for item in checks.values():
        if not isinstance(item, dict):
            raise ValueError('check must be an object')
        rc = item.get('rc')
        if rc is not None and type(rc) is not int:
            raise ValueError('rc must be an integer or null')
        if not all(isinstance(item.get(k), str) for k in ('out', 'error')):
            raise ValueError('out and error must be strings')
    return data


def summarize(data):
    d = validate(data)
    c = d['checks']
    status = c['status']
    state = 'unknown' if status['rc'] != 0 else (
        'dirty' if status['out'].strip() else 'clean')
    return {
        'kind': d['kind'], 'cwd': d['cwd'],
        'version': c['version']['out'].strip(),
        'git_root': c['root']['out'].strip() if c['root']['rc'] == 0 else None,
        'worktree_state': state,
        'failed_checks': [k for k, v in c.items() if v['rc'] != 0],
        'warning_checks': [k for k, v in c.items() if v['error'].strip()],
    }


def main():
    p = argparse.ArgumentParser(description=__doc__)
    modes = p.add_mutually_exclusive_group(required=True)
    modes.add_argument('--repo')
    modes.add_argument('--fixture', type=Path)
    args = p.parse_args()
    try:
        if args.fixture:
            if args.fixture.stat().st_size > 262144:
                raise ValueError('fixture exceeds 256 KiB')
            data = json.loads(args.fixture.read_text(encoding='utf-8'))
        else:
            data = collect(args.repo)
        result = {'evidence': validate(data), 'summary': summarize(data)}
    except (ValueError, OSError) as e:
        p.exit(2, f'input error: {e}\n')
    print(json.dumps(result, ensure_ascii=False, indent=2))


if __name__ == '__main__':
    main()

{
  "schema": 1,
  "kind": "synthetic_fixture",
  "cwd": "/example/demo",
  "checks": {
    "version": {
      "rc": 0,
      "out": "codex-cli FIXTURE-VERSION\n",
      "error": ""
    },
    "help": {
      "rc": 0,
      "out": "",
      "error": ""
    },
    "resume": {
      "rc": 0,
      "out": "",
      "error": ""
    },
    "fork": {
      "rc": 0,
      "out": "",
      "error": ""
    },
    "review": {
      "rc": 0,
      "out": "",
      "error": ""
    },
    "root": {
      "rc": 0,
      "out": "/example/demo\n",
      "error": ""
    },
    "branch": {
      "rc": 0,
      "out": "example-branch\n",
      "error": ""
    },
    "status": {
      "rc": 0,
      "out": "",
      "error": ""
    }
  }
}

import copy
import importlib.util
import json
from pathlib import Path
import subprocess
import sys
import tempfile
import unittest
from unittest.mock import patch

ROOT = Path(__file__).resolve().parents[1]
spec = importlib.util.spec_from_file_location('probe', ROOT/'scripts/cli_probe.py')
probe = importlib.util.module_from_spec(spec)
spec.loader.exec_module(probe)


class ProbeTests(unittest.TestCase):
    def setUp(self):
        self.data = json.loads((ROOT/'fixtures/clean.json').read_text())

    def test_synthetic_label_and_clean(self):
        r = probe.summarize(self.data)
        self.assertEqual(r['kind'], 'synthetic_fixture')
        self.assertEqual(r['worktree_state'], 'clean')

    def test_dirty_includes_untracked(self):
        self.data['checks']['status']['out'] = '?? new.txt\n'
        self.assertEqual(probe.summarize(self.data)['worktree_state'], 'dirty')

    def test_failed_status_is_unknown(self):
        self.data['checks']['status']['rc'] = 128
        self.assertEqual(probe.summarize(self.data)['worktree_state'], 'unknown')

    def test_missing_executable_is_not_clean(self):
        self.data['checks']['status'] = probe.run_fixed(None, [], '.')
        r = probe.summarize(self.data)
        self.assertEqual(r['worktree_state'], 'unknown')
        self.assertIn('status', r['failed_checks'])

    def test_warning_is_preserved(self):
        self.data['checks']['help']['error'] = 'demonstration warning'
        self.assertIn('help', probe.summarize(self.data)['warning_checks'])

    def test_boolean_exit_code_is_rejected(self):
        self.data['checks']['status']['rc'] = False
        with self.assertRaises(ValueError):
            probe.validate(self.data)

    def test_missing_and_unknown_keys_rejected(self):
        for key in ('status', 'arbitrary_shell'):
            d = copy.deepcopy(self.data)
            if key == 'status':
                del d['checks'][key]
            else:
                d['checks'][key] = {'rc': 0, 'out': '', 'error': ''}
            with self.assertRaises(ValueError):
                probe.validate(d)

    def test_timeout_is_explicit(self):
        with patch.object(probe.subprocess, 'run', side_effect=
                          subprocess.TimeoutExpired('codex', 10)):
            r = probe.run_fixed('codex', ['--help'], '.')
        self.assertIsNone(r['rc'])
        self.assertEqual(r['error'], 'timeout')

    def check_invalid_utf8(self, stream):
        # A tiny local Python child, not Git, Codex, or a model call.
        r = probe.run_fixed(
            sys.executable,
            ['-c', f'import sys; sys.{stream}.buffer.write(bytes([255]))'],
            ROOT,
        )
        self.assertIsNone(r['rc'])
        self.assertEqual(r['out'], '')
        self.assertEqual(r['error'],
                         'invalid UTF-8 command output; result unavailable')
        self.data['checks']['status'] = r
        summary = probe.summarize(self.data)
        self.assertEqual(summary['kind'], 'synthetic_fixture')
        self.assertEqual(summary['worktree_state'], 'unknown')
        self.assertIn('status', summary['failed_checks'])
        self.assertIn('status', summary['warning_checks'])

    def test_invalid_utf8_stdout_is_unknown(self):
        self.check_invalid_utf8('stdout')

    def test_invalid_utf8_stderr_is_unknown(self):
        self.check_invalid_utf8('stderr')

    def test_utf8_output_and_exit_code_are_preserved(self):
        source = ("import sys; "
                  "sys.stdout.buffer.write(bytes.fromhex('e4b8ade696870a')); "
                  "sys.stderr.buffer.write(bytes.fromhex('e8ada6e5918a0a')); "
                  "sys.exit(3)")
        r = probe.run_fixed(sys.executable, ['-c', source], ROOT)
        self.assertEqual(r, {'rc': 3, 'out': '中文\n', 'error': '警告\n'})

    def test_fixture_never_runs_subprocess(self):
        self.data['checks']['version']['out'] = '$(touch do-not-create)'
        with patch.object(probe.subprocess, 'run') as child:
            probe.summarize(self.data)
            child.assert_not_called()

    def test_invalid_directory(self):
        with tempfile.TemporaryDirectory() as tmp:
            file = Path(tmp)/'file.txt'
            file.write_text('fixture')
            with self.assertRaises(ValueError):
                probe.collect(file)

    def test_cli_fixture(self):
        result = subprocess.run(
            ['python3', str(ROOT/'scripts/cli_probe.py'), '--fixture',
             str(ROOT/'fixtures/clean.json')],
            capture_output=True, text=True, check=True,
        )
        self.assertEqual(json.loads(result.stdout)['summary']
                         ['worktree_state'], 'clean')


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

材料类型：完全合成的可见现象记录。时间、任务代号和动作均为教学编排。
这不是 Codex 导出的日志格式，也不是本次真实运行记录。

任务代号：DEMO-QUEUE（虚构）；原任务窗口仍保留。
时间基准：UTC；以下时间是合成样例。
2026-10-12T09:20:00Z  可见提示包含：
Error running remote compact task: stream disconnected before completion: Transport error: network error: error decoding response body
2026-10-12T09:22:00Z  用户另发了一次继续请求。
2026-10-12T09:23:00Z  该请求再次显示相同末端错误。

人工继续请求：1 次；客户端内部重试次数：未采集。
最后确认的任务成果：第一步计数函数；是否在故障轮中又改了文件：未知，需查工作区。
故障所属阶段：可见文字提及 remote compact；更细阶段未知。
整轮时长：未采集；单个请求耗时与服务器处理时间：未知。
客户端入口、应用构建、CLI/运行时版本、系统与架构、模型：未采集。
其他短对话是否正常：未测；网络/代理对照：未测；根因：未知。
未收集：账号、令牌、会话数据库、完整对话、原始请求体和真实项目文件。

材料类型：合成任务的完整交接；下面的 Git、文件与测试结果来自本次离线演练。
目标：给虚构队列统计 ready 项；保留 pending 分类，不重做已有工作。

当前状态
分支：handoff-fixture
基线提交（仅合成仓库）：a412074a21208137db581caf78173d918a0eb20e
第一步：精确匹配 ready 已实现；空列表返回 0。
第二步：尚未实现。['READY', 'pending'] 应得到 1。
先增加大小写用例，再做最小修改；不要把第二步记为已验收。

已有改动的归属（由本例作者预先指定，不能靠 Git 猜作者）
README.md：暂存的使用者备注，必须保留 pending 分类。
counter.py：未暂存的第一步实现。
next case.txt：未跟踪的第二步需求，必须连同内容一起保留。
test_counter.py：基线中已有的两个第一步测试。

当前四文件清单
README.md | 63 bytes | sha256 597de4eae06d3f660f30879bcb9e6cc8ad23ff3d79b68ed464342e0264a5c109
counter.py | 78 bytes | sha256 b626314c4c96178be9f3ba5e269f3f1fc550dd9c7471a87e9e3ebe88cb7ff4be
test_counter.py | 258 bytes | sha256 9b6f36d1e34bdab87bd31b09c99727dce208378746d5c67f56dfdd6771a25219
next case.txt | 147 bytes | sha256 cd874f76c3e7f61bf79aa267e54a4e33142e373f53058b8b2368f97d92ae442b

原目录和独立接收目录均执行：python3 -B -m unittest -v test_counter
两个测试：test_empty、test_ready_only；两处退出码均为 0。
验证环境：Linux、Python 3.12.14、git version 2.52.0。
第二步验收测试未运行；Codex、模型与压缩调用未运行。

接手顺序
1. 核对四个明确文件的字节数与 SHA-256；缺文件或哈希变更就停下重新确认。
2. 阅读 README.md 的已有备注和 next case.txt 的未完成需求。
3. 在允许执行测试的目录重跑上述两个测试，记录当前结果。
4. 先解释第二步准备改什么，确认修改范围后再实施。

边界
这里复制的是四个合成文件。Git 历史、暂存状态、忽略文件与其他项目文件没有被复制。
状态、diff、哈希清单是证据索引，不是完整备份。接收目录没有旧对话历史。
真实任务需由你逐项确认要保留的文件与允许的接收方，不能套用这四个文件名。
运行时压缩故障：本例未复现、未修复；交接可核对不代表原会话已恢复。

材料类型：完全合成的支持报告草稿，未发送。与真实 Codex 运行无关。
标题：长任务显示 remote compact 解码错误后，继续请求再次失败

发生时间：合成时间 2026-10-12 09:20–09:23 UTC。
客户端入口/应用构建：未采集。CLI 与运行时版本：未采集，不用另一安装的版本代替。
操作系统/架构：未采集。模型与推理档位：未采集。
可见错误：见同包 incident.synthetic.txt 的一条完整错误；公开时仅保留必要片段。

动作顺序
已有任务执行到第一步；可见文字提及 remote compact 后出现解码失败。
用户另发一次继续请求，又得到相同末端错误。
预期：支持安全继续，或告知如何从保留的工作状态恢复。
实际：原任务不能确认已继续；内部自动重试次数与请求耗时未知。

范围与对照
其他短对话：未测。其他长对话：未测。其他网络：未测。
是否特定模型/平台独有：未知。是否由服务、网络或客户端引起：未知。
已完成动作：整理可见文字与待核对的工作清单，保留原任务窗口。
未做动作：未修改压缩阈值、传输配置或历史；未进行付费复现；未上传诊断。
需要支持方确认：与上述客户端版本匹配的受支持恢复办法，以及最小诊断字段。

公开前人工审阅
去除真实项目名、源码、路径、账号、邮箱、令牌、私有提示词与完整日志。
任务/请求/反馈标识仅在官方私密支持渠道确有需要时另行提供，不贴到公开 issue。
工作交接包与诊断报告分开审阅，不把项目文件顺带公开。
这个草稿没有已知根因或通用修复结论，不应原样提交为真实故障。

Codex 压缩/解码故障记录（填写真实观察，未采集项保留未知）
时间与时区：
客户端入口及其应用构建：
实际受影响 CLI/运行时版本（来源）：
系统与架构：
模型/推理档位（可见时）：
可见错误的最小脱敏片段：
错误前最后一个已确认完成的动作：
人工继续请求次数：
客户端可见自动重试次数（没有看到则未知）：
最后是否明确完成/失败/仍在运行：
是否还有写入或后台任务（未确认则未知）：
其他短对话/长对话的已知观察（不要为了填表重复付费调用）：
工作区检查的时间、退出码、文件清单及未完成项（单独本地保存）：
期望行为与实际行为：
已尝试动作及结果：
未测项与未知根因：
公开/私密接收渠道及敏感信息审阅：

"""Offline teaching fixture only. No Codex, network, or user repository access."""
import hashlib
import json
import os
from pathlib import Path
import platform
import shutil
import subprocess
import sys
import tempfile

FILES = {
    "README.md": "Synthetic queue counter.\nUser note: keep the pending category.\n",
    "counter.py": "def count_ready(states):\n    return sum(state == 'ready' for state in states)\n",
    "test_counter.py": "import unittest\nfrom counter import count_ready\n\nclass StepOne(unittest.TestCase):\n    def test_empty(self):\n        self.assertEqual(count_ready([]), 0)\n\n    def test_ready_only(self):\n        self.assertEqual(count_ready(['ready', 'pending', 'ready']), 2)\n",
    "next case.txt": "Step 2, NOT IMPLEMENTED: count_ready(['READY', 'pending']) should return 1.\nFirst add a case-insensitive test; preserve existing pending behavior.\n",
}
EXPECTED_STATUS = 'M  README.md\n M counter.py\n?? "next case.txt"\n'


def require(ok, message):
    if not ok:
        raise ValueError(message)


def sha(data):
    return hashlib.sha256(data).hexdigest()


def manifest(folder):
    result = {}
    for name in FILES:
        p = folder / name
        require(not p.is_symlink() and p.is_file(), "missing/nonregular: " + name)
        data = p.read_bytes()
        result[name] = {"bytes": len(data), "sha256": sha(data)}
    return result


def verify(folder, expected):
    require(set(expected) == set(FILES), "manifest scope changed")
    actual = manifest(folder)
    require(actual == expected, "file bytes changed")
    return True


def run(argv, cwd, env):
    return subprocess.run(argv, cwd=cwd, env=env, capture_output=True,
                          text=True, encoding="utf-8", timeout=20, check=True)


def run_demo():
    git = shutil.which("git")
    require(git is not None, "Git is required; nothing will be installed")
    with tempfile.TemporaryDirectory(prefix="compact-handoff-") as temp:
        root = Path(temp)
        repo, receiver, home, empty = [root / n for n in
                                        ("demo repo", "receiver", "home", "empty")]
        for p in (repo, receiver, home, empty):
            p.mkdir()
        env = {"PATH": os.environ.get("PATH", os.defpath), "HOME": str(home),
               "LC_ALL": "C", "GIT_CONFIG_NOSYSTEM": "1",
               "GIT_CONFIG_GLOBAL": os.devnull, "GIT_OPTIONAL_LOCKS": "0",
               "GIT_TERMINAL_PROMPT": "0", "GIT_ATTR_NOSYSTEM": "1",
               "GIT_AUTHOR_DATE": "2026-10-13T00:00:00+0000",
               "GIT_COMMITTER_DATE": "2026-10-13T00:00:00+0000"}
        # Explicitly empty environment configuration and hooks for this fixture.
        prefix = [git, "-c", "core.fsmonitor=false", "-c", "core.autocrlf=false",
                  "-c", "core.hooksPath=" + str(empty), "-c", "commit.gpgSign=false",
                  "-c", "core.attributesFile=" + os.devnull]

        def g(*args):
            return run(prefix + list(args), repo, env).stdout

        g("init", "--template=" + str(empty), "--initial-branch=handoff-fixture")
        g("config", "user.name", "Synthetic Fixture")
        g("config", "user.email", "fixture@example.invalid")
        baseline = {"README.md": "Synthetic queue counter.\n",
                    "counter.py": "def count_ready(states):\n    return 0\n",
                    "test_counter.py": FILES["test_counter.py"]}
        for name, content in baseline.items():
            (repo / name).write_text(content, encoding="utf-8", newline="\n")
        g("add", "--", *baseline)
        g("commit", "-m", "synthetic baseline")
        for name, content in FILES.items():
            (repo / name).write_text(content, encoding="utf-8", newline="\n")
        g("add", "--", "README.md")

        def snapshot():
            return {"files": manifest(repo), "index_sha256": sha((repo / ".git/index").read_bytes()),
                    "head": g("rev-parse", "HEAD").strip(),
                    "branch": g("branch", "--show-current").strip(),
                    "status": g("status", "--porcelain=v1", "--untracked-files=all"),
                    "unstaged_diff": g("diff", "--no-ext-diff", "--no-textconv"),
                    "staged_diff": g("diff", "--cached", "--no-ext-diff", "--no-textconv")}

        before = snapshot()
        require(before["status"] == EXPECTED_STATUS, "unexpected fixture status")
        test_args = [sys.executable, "-B", "-m", "unittest", "-v", "test_counter"]
        original_test = run(test_args, repo, env)
        for name in FILES:
            (receiver / name).write_bytes((repo / name).read_bytes())
        verify(receiver, before["files"])
        copied_test = run(test_args, receiver, env)
        after = snapshot()
        require(before == after, "source fixture changed during collection/test/copy")
        require("next case.txt" not in before["unstaged_diff"] + before["staged_diff"],
                "unexpected untracked content in tracked diff")
        return {"kind": "synthetic_workspace_handoff", "codex_run": False,
                "runtime_issue": "not_reproduced_or_fixed",
                "environment": {"python": platform.python_version(),
                                "git": run([git, "--version"], repo, env).stdout.strip(),
                                "system": platform.system()},
                "before": before, "after": after,
                "checks": {"three_work_states_present": True,
                           "source_bytes_index_head_status_diffs_unchanged": before == after,
                           "receiver_hashes_match": True,
                           "original_step1_tests_exit": original_test.returncode,
                           "receiver_step1_tests_exit": copied_test.returncode,
                           "untracked_is_absent_from_both_diffs": True},
                "step1_test_names": ["test_empty", "test_ready_only"],
                "original_test_output": original_test.stdout + original_test.stderr,
                "receiver_test_output": copied_test.stdout + copied_test.stderr,
                "step2": {"state": "not_implemented", "instruction": FILES["next case.txt"],
                          "acceptance_test_run": False},
                "not_preserved_by_copy": ["Git history", "index staging state",
                                          "ignored files", "other project files"],
                "scope": "four listed synthetic files only"}


if __name__ == "__main__":
    try:
        print(json.dumps(run_demo(), ensure_ascii=False, indent=2))
    except (ValueError, OSError, subprocess.SubprocessError) as e:
        sys.exit("demo failed: " + str(e))

"""Independent negative and replay checks for the teaching handoff."""
import copy
from pathlib import Path
import tempfile
import unittest

import demo_handoff as demo


class HandoffChecks(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.addCleanup(self.temp.cleanup)
        self.folder = Path(self.temp.name)
        for name, content in demo.FILES.items():
            (self.folder / name).write_bytes(content.encode("utf-8"))
        self.expected = demo.manifest(self.folder)

    def test_complete_copy(self):
        self.assertTrue(demo.verify(self.folder, self.expected))

    def test_changed_tracked_file(self):
        (self.folder / "counter.py").write_bytes(b"changed\n")
        with self.assertRaisesRegex(ValueError, "file bytes changed"):
            demo.verify(self.folder, self.expected)

    def test_missing_untracked_file(self):
        (self.folder / "next case.txt").unlink()
        with self.assertRaisesRegex(ValueError, "missing/nonregular"):
            demo.verify(self.folder, self.expected)

    def test_changed_preserved_user_note(self):
        (self.folder / "README.md").write_bytes(b"note lost\n")
        with self.assertRaises(ValueError):
            demo.verify(self.folder, self.expected)

    def test_missing_manifest_entry(self):
        del self.expected["next case.txt"]
        with self.assertRaisesRegex(ValueError, "manifest scope changed"):
            demo.verify(self.folder, self.expected)

    def test_unexpected_manifest_path(self):
        self.expected["../outside"] = copy.deepcopy(self.expected["README.md"])
        with self.assertRaisesRegex(ValueError, "manifest scope changed"):
            demo.verify(self.folder, self.expected)

    def test_incorrect_byte_count(self):
        self.expected["README.md"]["bytes"] += 1
        with self.assertRaises(ValueError):
            demo.verify(self.folder, self.expected)

    def test_symlink_rejected(self):
        original = self.folder / "README.md"
        original.rename(self.folder / "note-copy")
        original.symlink_to("note-copy")
        with self.assertRaisesRegex(ValueError, "missing/nonregular"):
            demo.verify(self.folder, self.expected)

    def test_empty_file_is_not_original(self):
        (self.folder / "counter.py").write_bytes(b"")
        with self.assertRaises(ValueError):
            demo.verify(self.folder, self.expected)

    def test_unknown_runtime_cannot_become_fixed(self):
        report = demo.run_demo()
        self.assertFalse(report["codex_run"])
        self.assertEqual(report["runtime_issue"], "not_reproduced_or_fixed")
        self.assertEqual(report["before"], report["after"])
        self.assertEqual(report["before"]["status"], demo.EXPECTED_STATUS)
        self.assertIn("User note: keep the pending category.", report["before"]["staged_diff"])
        self.assertIn("sum(state == 'ready'", report["before"]["unstaged_diff"])
        self.assertNotIn("next case.txt", report["before"]["unstaged_diff"])
        self.assertEqual(report["checks"]["receiver_step1_tests_exit"], 0)
        self.assertEqual(report["step2"]["state"], "not_implemented")
        self.assertFalse(report["step2"]["acceptance_test_run"])


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