274 行
8.1 KiB
Python
可执行文件
274 行
8.1 KiB
Python
可执行文件
#!/usr/bin/env python3
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from textwrap import shorten
|
|
|
|
|
|
IGNORE_PREFIXES = (
|
|
"# AGENTS.md instructions",
|
|
"<environment_context>",
|
|
"<turn_aborted>",
|
|
)
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description="Generate a Codex handoff summary for a project.")
|
|
parser.add_argument("--project-dir", required=True, help="Target project directory")
|
|
parser.add_argument("--output", required=True, help="Output markdown path")
|
|
parser.add_argument(
|
|
"--sessions-root",
|
|
default=str(Path.home() / ".codex" / "sessions"),
|
|
help="Codex sessions root directory",
|
|
)
|
|
parser.add_argument("--limit", type=int, default=4, help="Number of recent turns to include")
|
|
return parser.parse_args()
|
|
|
|
|
|
def normalize_path(path: str | Path) -> Path:
|
|
return Path(path).expanduser().resolve()
|
|
|
|
|
|
def read_first_json_line(path: Path) -> dict | None:
|
|
try:
|
|
with path.open("r", encoding="utf-8") as handle:
|
|
line = handle.readline()
|
|
return json.loads(line) if line else None
|
|
except (OSError, json.JSONDecodeError):
|
|
return None
|
|
|
|
|
|
def find_latest_session(project_dir: Path, sessions_root: Path) -> Path | None:
|
|
if not sessions_root.exists():
|
|
return None
|
|
|
|
candidates: list[tuple[float, Path]] = []
|
|
for session_file in sessions_root.rglob("*.jsonl"):
|
|
first = read_first_json_line(session_file)
|
|
if not first:
|
|
continue
|
|
cwd = first.get("payload", {}).get("cwd")
|
|
if not cwd:
|
|
continue
|
|
try:
|
|
if normalize_path(cwd) == project_dir:
|
|
candidates.append((session_file.stat().st_mtime, session_file))
|
|
except OSError:
|
|
continue
|
|
|
|
if not candidates:
|
|
return None
|
|
|
|
candidates.sort(key=lambda item: item[0], reverse=True)
|
|
return candidates[0][1]
|
|
|
|
|
|
def extract_message_text(payload: dict) -> str:
|
|
parts = []
|
|
for item in payload.get("content", []):
|
|
if item.get("type") in {"input_text", "output_text"}:
|
|
text = (item.get("text") or "").strip()
|
|
if text:
|
|
parts.append(text)
|
|
return "\n\n".join(parts).strip()
|
|
|
|
|
|
def is_substantive(text: str) -> bool:
|
|
if not text:
|
|
return False
|
|
stripped = text.strip()
|
|
if any(stripped.startswith(prefix) for prefix in IGNORE_PREFIXES):
|
|
return False
|
|
if stripped.startswith("<environment_context>") or stripped.startswith("<turn_aborted>"):
|
|
return False
|
|
return True
|
|
|
|
|
|
def compact(text: str, width: int = 220) -> str:
|
|
return shorten(" ".join(text.split()), width=width, placeholder="...")
|
|
|
|
|
|
def parse_session(session_path: Path, limit: int) -> tuple[list[str], list[str]]:
|
|
user_turns: list[str] = []
|
|
assistant_turns: list[str] = []
|
|
|
|
with session_path.open("r", encoding="utf-8") as handle:
|
|
for raw_line in handle:
|
|
try:
|
|
obj = json.loads(raw_line)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
|
|
if obj.get("type") != "response_item":
|
|
continue
|
|
|
|
payload = obj.get("payload", {})
|
|
if payload.get("type") != "message":
|
|
continue
|
|
|
|
role = payload.get("role")
|
|
text = extract_message_text(payload)
|
|
if not is_substantive(text):
|
|
continue
|
|
|
|
summary = compact(text, 300 if role == "user" else 220)
|
|
if role == "user":
|
|
user_turns.append(summary)
|
|
elif role == "assistant":
|
|
assistant_turns.append(summary)
|
|
|
|
return user_turns[-limit:], assistant_turns[-limit:]
|
|
|
|
|
|
def run_git(project_dir: Path, *args: str) -> str:
|
|
result = subprocess.run(
|
|
["git", *args],
|
|
cwd=str(project_dir),
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
if result.returncode != 0:
|
|
return ""
|
|
return result.stdout.strip()
|
|
|
|
|
|
def repo_state(project_dir: Path) -> dict[str, str]:
|
|
return {
|
|
"branch": run_git(project_dir, "rev-parse", "--abbrev-ref", "HEAD"),
|
|
"status": run_git(project_dir, "status", "--short", "--branch"),
|
|
"recent_commits": run_git(project_dir, "log", "-3", "--oneline", "--decorate", "--no-color"),
|
|
}
|
|
|
|
|
|
def render_with_session(project_dir: Path, session_path: Path, limit: int) -> str:
|
|
user_turns, assistant_turns = parse_session(session_path, limit)
|
|
state = repo_state(project_dir)
|
|
goal = user_turns[-1] if user_turns else "Continue the latest Codex task for this workspace."
|
|
|
|
lines = [
|
|
"# Codex Handoff",
|
|
"",
|
|
f"- Generated at: {datetime.now(timezone.utc).isoformat()}",
|
|
f"- Project root: `{project_dir}`",
|
|
f"- Session file: `{session_path}`",
|
|
"",
|
|
"## Goal",
|
|
"",
|
|
goal,
|
|
"",
|
|
"## Current State",
|
|
"",
|
|
f"- Branch: `{state['branch'] or 'unknown'}`",
|
|
"- Git status:",
|
|
"```text",
|
|
state["status"] or "(clean or unavailable)",
|
|
"```",
|
|
"- Recent commits:",
|
|
"```text",
|
|
state["recent_commits"] or "(no commits yet)",
|
|
"```",
|
|
"",
|
|
"## Recent User Turns",
|
|
"",
|
|
]
|
|
|
|
if user_turns:
|
|
lines.extend(f"- {turn}" for turn in user_turns)
|
|
else:
|
|
lines.append("- No substantive user turns were found in the matching session.")
|
|
|
|
lines.extend(
|
|
[
|
|
"",
|
|
"## Recent Assistant Turns",
|
|
"",
|
|
]
|
|
)
|
|
|
|
if assistant_turns:
|
|
lines.extend(f"- {turn}" for turn in assistant_turns)
|
|
else:
|
|
lines.append("- No substantive assistant turns were found in the matching session.")
|
|
|
|
next_action = (
|
|
"Open the project, inspect `.codex-sync/runtime/status.json` and `git status -sb`, then continue from the latest user request above."
|
|
)
|
|
if state["status"] and state["status"] != "## HEAD (no branch)":
|
|
next_action = (
|
|
"Read `.codex-sync/runtime/status.json`, inspect the git status block above, and continue from the latest user request while preserving any uncommitted work."
|
|
)
|
|
|
|
lines.extend(
|
|
[
|
|
"",
|
|
"## Next Action",
|
|
"",
|
|
next_action,
|
|
"",
|
|
]
|
|
)
|
|
return "\n".join(lines)
|
|
|
|
|
|
def render_fallback(project_dir: Path) -> str:
|
|
state = repo_state(project_dir)
|
|
lines = [
|
|
"# Codex Handoff",
|
|
"",
|
|
f"- Generated at: {datetime.now(timezone.utc).isoformat()}",
|
|
f"- Project root: `{project_dir}`",
|
|
"- Session file: `(no matching Codex session found)`",
|
|
"",
|
|
"## Goal",
|
|
"",
|
|
"Continue this repository from its current git state. No matching local Codex session was found for this workspace, so treat the repository contents and recent commits as the source of truth.",
|
|
"",
|
|
"## Current State",
|
|
"",
|
|
f"- Branch: `{state['branch'] or 'unknown'}`",
|
|
"- Git status:",
|
|
"```text",
|
|
state["status"] or "(clean or unavailable)",
|
|
"```",
|
|
"- Recent commits:",
|
|
"```text",
|
|
state["recent_commits"] or "(no commits yet)",
|
|
"```",
|
|
"",
|
|
"## Next Action",
|
|
"",
|
|
"Inspect the repository, read `.codex-sync/runtime/status.json`, and decide the next concrete implementation step from the current files and recent commits.",
|
|
"",
|
|
]
|
|
return "\n".join(lines)
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
project_dir = normalize_path(args.project_dir)
|
|
output_path = normalize_path(args.output)
|
|
sessions_root = normalize_path(args.sessions_root)
|
|
|
|
output_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
session_path = find_latest_session(project_dir, sessions_root)
|
|
if session_path:
|
|
content = render_with_session(project_dir, session_path, args.limit)
|
|
else:
|
|
content = render_fallback(project_dir)
|
|
|
|
output_path.write_text(content, encoding="utf-8")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|