
Codex Sdk
- 5 installs
- 5 repo stars
- Updated August 5, 2026
- bjornmelin/dev-skills
codex-sdk is a skill giving architect-level guidance and scripts for building agentic coding systems with the OpenAI Codex SDK and CLI.
About
codex-sdk gives architect-level guidance, workflows, and scripts for building agentic coding systems with OpenAI Codex. A developer uses it for the Codex SDK (@openai/codex-sdk) threads and streaming, Codex CLI automation (codex exec), MCP server usage, multi-agent orchestration via the OpenAI Agents SDK, and durable state in SQLite. It emphasizes structured outputs, audit trails, and safe-by-default sandbox and approval patterns.
- Architect-level guidance for building agentic coding systems with the OpenAI Codex SDK and CLI
- Covers codex exec JSONL, @openai/codex-sdk threads/streaming, and codex mcp-server orchestration
- Uses SQLite for durable state, audit logs, and resumable runs with safe-by-default sandboxing
Codex Sdk by the numbers
- 5 all-time installs (skills.sh)
- Ranked #13,046 of 16,546 AI & Agent Building skills by installs in the Skillselion catalog
- Data as of Aug 5, 2026 (Skillselion catalog sync)
codex-sdk capabilities & compatibility
- Capabilities
- agent orchestration · codex automation · mcp server · durable state · audit logging
- Works with
- openai
- Use cases
- orchestration · memory · planning · code review
- Pricing
- Bring your own API key
What codex-sdk says it does
Build reliable, auditable, multi-step coding workflows that scale from a single run to multi-agent orchestration.
**Need a scriptable one-shot in CI or cron?** → use `codex exec`
SQLite is the simplest reliable substrate for:
npx skills add https://github.com/bjornmelin/dev-skills --skill codex-sdkAdd your badge
Show developers this skill is listed on Skillselion. Paste this into your README.
| Installs | 5 |
|---|---|
| repo stars | ★ 5 |
| Last updated | August 5, 2026 |
| Repository | bjornmelin/dev-skills ↗ |
What it does
Build reliable multi-step and multi-agent coding workflows with the OpenAI Codex SDK and CLI.
Who is it for?
Building multi-step, auditable, and multi-agent coding workflows with Codex.
Skip if: Simple one-off prompts where full orchestration and durable state are unnecessary.
When should I use this skill?
Building agentic coding systems with the Codex SDK, automating codex exec, running codex mcp-server, or orchestrating multi-agent workflows.
What you get
Reliable, auditable multi-step Codex workflows with JSONL-to-SQLite audit logs and gated multi-agent handoffs.
- ExecPlans for durable planning
- JSONL-to-SQLite audit pipeline
- multi-agent orchestration shape
By the numbers
- 4-branch workflow decision tree
- 6-step default safe workflow
- 12 reference files
Files
Codex SDK + Codex CLI (master skill)
Build reliable, auditable, multi-step coding workflows that scale from a single run to multi-agent orchestration.
ExecPlans (durable planning)
For multi-hour work, keep intent durable across compaction/restarts using an ExecPlan:
- Contract:
references/execplans.md - Template + rules:
.agent/PLANS.md(usescripts/init_agent_workspace.pyto bootstrap a repo) - ExecPlan skeleton:
assets/templates/execplan.md(or generate a file withscripts/new_execplan.py)
Workflow decision tree
1. Need a scriptable one-shot in CI or cron? → use codex exec (references/codex-cli-exec.md) 2. Need a server-side app controlling Codex programmatically? → use @openai/codex-sdk (references/codex-sdk-typescript.md) 3. Need multi-agent orchestration or dynamic tools? → run codex mcp-server and orchestrate via OpenAI Agents SDK (references/mcp-and-agents-sdk.md) 4. Need durability across runs (memory, caching, resumable state)? → persist run metadata and event logs in SQLite (references/state-memory-sqlite.md)
Default workflow (safe + production-friendly)
1. Inventory inputs (repo root, diffs, failing commands, constraints). 2. Choose sandbox + approvals (least privilege; default to read-only). 3. Plan in explicit steps (short, verifiable; include stop conditions). 4. Use structured outputs (JSON Schema) and validate before acting. 5. Record an audit trail (JSONL events → SQLite). 6. Verify (re-run tests/lint/format; stop).
Durable state and “memory” (SQLite)
SQLite is the simplest reliable substrate for:
- recording every
codex exec --jsonevent as an immutable audit log - indexing runs by repo, branch, and purpose
- storing
threadIdso runs can resume deterministically - caching expensive analysis artifacts (diff summaries, inventories, dependency graphs)
Use the bundled scripts
- Initialize a DB:
python3 scripts/codex_jsonl_to_sqlite.py --db codex-runs.sqlite --init- Ingest a JSONL run:
codex exec --json "<prompt>" | python3 scripts/codex_jsonl_to_sqlite.py --db codex-runs.sqlite --run-label "ci-autofix"- Summarize runs:
python3 scripts/codex_sqlite_report.py --db codex-runs.sqlite --latest
Multi-agent orchestration (recommended shape)
Use:
- a single orchestrator responsible for gating and artifact checks
- multiple scoped worker agents with strict deliverables
- structured outputs at boundaries (handoff payloads, review findings, test results)
- traces/telemetry to debug and tune
Details: references/mcp-and-agents-sdk.md
Safety and policy
Prefer:
- analysis-only:
sandbox: read-only,approval-policy: never - controlled edits:
sandbox: workspace-write,approval-policy: on-request/on-failure - block risky commands with
execpolicyrules (references/safety-and-execpolicy.md)
Resources
references/
references/codex-sdk-typescript.md– SDK patterns (threads, streaming, schemas)references/codex-cli-exec.md– CLI patterns (JSONL, schema files, resume)references/mcp-and-agents-sdk.md– Codex as MCP server + multi-agent orchestrationreferences/agents-sdk-consistent-workflows.md– gated handoffs + traces with Codex MCP + Agents SDKreferences/execplans.md– ExecPlans for long-running work across compactionreferences/state-memory-sqlite.md– SQLite schema + memory/caching patternsreferences/safety-and-execpolicy.md– sandboxing, approvals, prompt-injection defensesreferences/codex-config-knobs.md– config keys and feature flags that matterreferences/orchestration-patterns.md– planner/executor/verifier and orchestrator/worker patternsreferences/rag-and-memory.md– SQLite-first shared memory and RAG guidancereferences/context-personalization.md– state + memory notes personalization patterns (Agents SDK)
scripts/
scripts/codex_jsonl_to_sqlite.py– ingest Codex JSONL into SQLitescripts/codex_sqlite_report.py– summarize runs from SQLitescripts/init_agent_workspace.py– create.agent/AGENTS.md+.agent/PLANS.mdfrom templatesscripts/new_execplan.py– generateexecplans/execplan-*.mdfrom the ExecPlan template
assets/
assets/templates/– copy/paste templates (ExecPlan, prompts, schemas)assets/templates/agents-sdk/– Agents SDK starter snippets (MCP stdio, sessions, personalization)
.agent/AGENTS.md
This folder defines how agentic work is planned, executed, and verified in this repository.
Working agreements
- Prefer structured outputs for decisions that affect code.
- Default to read-only analysis unless edits are required.
- Keep changes small and verifiable; run checks before declaring done.
ExecPlans (planning contract)
- Use
.agent/PLANS.mdas the canonical planning standard for multi-step work. - For any task likely to outlive a single session (multi-hour work, migrations, large refactors, multi-agent workflows), create an ExecPlan under
execplans/and keep it updated. - When resuming after context compaction or a new session, re-open the ExecPlan and continue from its Progress section (do not rely on chat history).
- The orchestrator (human or agent) owns plan correctness; workers should only add scoped evidence and never expand scope unilaterally.
.agent/PLANS.md (ExecPlans)
This file defines the planning standard for ExecPlans: living, self-contained design + execution documents for work that can’t reliably fit in a single chat/session.
ExecPlans exist to survive:
- context window limits,
- conversation compaction,
- agent restarts,
- multi-agent handoffs.
When to create an ExecPlan
Create an ExecPlan when the task is any of the following:
- multi-hour work or multi-PR work,
- migrations, refactors, or “touch many files” changes,
- high-risk changes (security, data loss, auth, destructive ops),
- anything where a future contributor must be able to resume from disk.
Non-negotiables
1. Self-contained
- Assume the reader has only a fresh repo checkout + this ExecPlan file.
- Include definitions for any non-obvious term you use.
- Include exact commands, file paths, and expected outputs.
2. Living document
- Update Progress, Surprises & Discoveries, and Decision Log continuously.
- Before stopping, record “what’s done” and “what’s next” in Progress.
3. Outcome-first
- Acceptance is phrased as observable behavior (commands + expected output), not internal implementation attributes.
4. Idempotent and safe
- Steps should be safe to re-run.
- If a step can fail halfway, include how to recover.
How to use ExecPlans
- Authoring: start from the template; fill in repo-specific context and make the plan executable by a novice.
- Executing: treat the ExecPlan as the single source of truth; proceed milestone-by-milestone without asking for “next steps”.
- Resuming: re-open the ExecPlan and continue from Progress; do not depend on chat history.
Formatting rules (avoid broken plans)
- Prefer prose; use checklists only in Progress.
- Avoid nested triple-backtick fences inside an ExecPlan. Use indentation for:
- commands
- transcripts
- code excerpts
- diffs
- If an ExecPlan is embedded inside another document, wrap it in a single fenced block labeled
md. - If an ExecPlan is the entire contents of its own
.mdfile, do not wrap it in triple backticks.
Where ExecPlans live
Store ExecPlans under execplans/:
execplans/execplan-<short-name>.md
ExecPlan template
Copy from assets/templates/execplan.md (or generate one with the skill script scripts/new_execplan.py).
import asyncio
import os
from agents import Agent, Runner
from agents.mcp import MCPServerStdio
from agents.tracing import trace
"""
Multi-agent workflow skeleton:
- Run Codex CLI as an MCP server (stdio)
- Orchestrate specialized agents via handoffs
- Enforce gated handoffs (file existence / checks)
- Keep it traceable
This is intentionally minimal: adapt scopes, gates, and deliverables to your project.
"""
async def main() -> None:
repo_root = os.getcwd()
async with MCPServerStdio(
name="Codex CLI",
params={"command": "npx", "args": ["-y", "codex", "mcp-server"]},
# Long workflows can take time; keep the MCP client session alive.
client_session_timeout_seconds=3600,
) as codex_server:
# Workers: keep scopes narrow. They should call the MCP tool `codex`/`codex-reply`
# with explicit safety args when they need to read/write.
designer = Agent(
name="Designer",
instructions=(
"Write a short design spec. Save it as design/design_spec.md.\n"
"When creating files, call Codex MCP with "
'{"approval-policy":"never","sandbox":"workspace-write","cwd":"."}.\n'
"When done, hand off back to the Project Manager."
),
mcp_servers=[codex_server],
)
developer = Agent(
name="Developer",
instructions=(
"Implement only what the design spec requires.\n"
"When creating files, call Codex MCP with "
'{"approval-policy":"never","sandbox":"workspace-write","cwd":"."}.\n'
"When done, hand off back to the Project Manager."
),
mcp_servers=[codex_server],
)
tester = Agent(
name="Tester",
instructions=(
"Write a minimal test plan and run the checks the PM requests.\n"
"When creating files, call Codex MCP with "
'{"approval-policy":"never","sandbox":"workspace-write","cwd":"."}.\n'
"When done, hand off back to the Project Manager."
),
mcp_servers=[codex_server],
)
pm = Agent(
name="Project Manager",
instructions=(
"You are the orchestrator.\n"
"Create an ExecPlan in execplans/ and keep it updated.\n"
"Do not advance to the next stage until gates are satisfied.\n"
"\n"
"Gates (example):\n"
"- After Designer: design/design_spec.md exists.\n"
"- After Developer: expected code artifacts exist.\n"
"- After Tester: test plan exists and checks pass.\n"
"\n"
"Handoff order:\n"
"1) Designer\n"
"2) Developer\n"
"3) Tester\n"
"\n"
"If a gate fails, ask the owning agent to fix only that failure.\n"
),
handoffs=[designer, developer, tester],
mcp_servers=[codex_server],
)
with trace("codex-mcp-multiagent", group_id="example-run"):
await Runner.run(
pm,
f"Repo root is {repo_root}. Build a tiny demo feature and validate it. Keep everything small.",
max_turns=30,
)
if __name__ == "__main__":
asyncio.run(main())
import asyncio
import datetime as dt
import json
import re
from dataclasses import dataclass, field
from agents import Agent, RunContextWrapper, Runner, SQLiteSession, function_tool
"""
Context personalization skeleton (memory notes):
- Keep a structured state object in code (source of truth)
- Capture candidate memories via a dedicated tool (distillation)
- Consolidate session notes into durable notes (dedupe + "latest wins")
- Inject only the relevant slice into agent instructions for the next run
This avoids relying on the raw transcript as your only memory store.
"""
@dataclass
class MemoryNote:
type: str # "preference" | "constraint" | "fact"
key: str
value: object
confidence: float
source: str # prefer "user"
created_at: str
@dataclass
class UserState:
profile: dict[str, object] = field(default_factory=dict)
global_notes: list[MemoryNote] = field(default_factory=list)
session_notes: list[MemoryNote] = field(default_factory=list)
def now_iso() -> str:
return dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds")
def looks_sensitive(text: str) -> bool:
# Minimal guardrail examples (extend for your domain).
if re.search(r"\b\d{3}-\d{2}-\d{4}\b", text): # SSN-like
return True
if re.search(r"\b(?:api[_-]?key|secret|password)\b", text, re.I):
return True
return False
@function_tool
async def save_memory_note(wrapper: RunContextWrapper[UserState], note_json: str) -> str:
"""
Save a candidate memory note as JSON.
Only store durable preferences/constraints; never store secrets.
"""
try:
obj = json.loads(note_json)
except json.JSONDecodeError:
return "Rejected: invalid JSON."
note_type = str(obj.get("type") or "")
key = str(obj.get("key") or "")
value = obj.get("value")
confidence = float(obj.get("confidence") or 0.5)
source = str(obj.get("source") or "user")
if note_type not in {"preference", "constraint", "fact"}:
return "Rejected: invalid type."
if not key:
return "Rejected: missing key."
serialized = json.dumps({"key": key, "value": value}, ensure_ascii=False)
if looks_sensitive(serialized):
return "Rejected: looks sensitive."
wrapper.context.session_notes.append(
MemoryNote(
type=note_type,
key=key,
value=value,
confidence=max(0.0, min(1.0, confidence)),
source=source,
created_at=now_iso(),
)
)
return "Saved."
def inject_memory(state: UserState, max_notes: int = 8) -> str:
# Keep injection small. Prefer top-K by recency and/or relevance (add relevance ranking as needed).
notes = state.global_notes[-max_notes:]
lines = [
"You are a helpful assistant.",
"",
"Personalization context (advisory):",
"<profile>",
json.dumps(state.profile, ensure_ascii=False),
"</profile>",
"<memories>",
]
for n in notes:
lines.append(f"- ({n.type}) {n.key}: {json.dumps(n.value, ensure_ascii=False)}")
lines += ["</memories>", "", "Precedence: current user input > session context > memories."]
return "\n".join(lines)
def consolidate(state: UserState) -> None:
# Simple “latest wins” per key.
by_key: dict[str, MemoryNote] = {n.key: n for n in state.global_notes}
for note in state.session_notes:
by_key[note.key] = note
state.global_notes = sorted(by_key.values(), key=lambda n: n.created_at)
state.session_notes = []
async def main() -> None:
state = UserState(profile={"locale": "en-US"})
session = SQLiteSession("user-123", "agent_history.sqlite")
# Run 1
agent = Agent[UserState](
name="Concierge",
instructions=inject_memory(state),
tools=[save_memory_note],
)
await Runner.run(agent, "I prefer vegetarian meals. Please remember that.", context=state, session=session)
consolidate(state)
# Run 2 (memory should be injected)
agent2 = Agent[UserState](
name="Concierge",
instructions=inject_memory(state),
tools=[save_memory_note],
)
result = await Runner.run(agent2, "Plan a dinner recommendation.", context=state, session=session)
print(result.final_output)
if __name__ == "__main__":
asyncio.run(main())
import path from "node:path";
import { Agent, run, MCPServerStdio } from "@openai/agents";
// Minimal MCP stdio example (TypeScript):
// - Connect to a local MCP server (filesystem shown here)
// - Attach it to an agent
// - Run a query that uses tools
const dataDir = path.join(process.cwd(), "data");
const filesystemServer = new MCPServerStdio({
name: "Filesystem MCP Server",
fullCommand: `npx -y @modelcontextprotocol/server-filesystem ${dataDir}`,
});
await filesystemServer.connect();
try {
const agent = new Agent({
name: "File Assistant",
instructions: "Use available tools to read files and answer questions.",
mcpServers: [filesystemServer],
});
const result = await run(agent, "List files under ./data and summarize markdown contents.");
console.log(result.finalOutput);
} finally {
await filesystemServer.close();
}
import { ConversationalAgent } from "@openai/agents";
import { MemorySession } from "@openai/agents/sessions/memory";
import type { AgentInputItem } from "@openai/agents";
// Sessions example (TypeScript):
// - Use a session to persist history
// - Use sessionInputCallback to deterministically trim what the model sees
const session = new MemorySession();
const agent = new ConversationalAgent({ model: "gpt-4o" });
const sessionInputCallback = async (history: AgentInputItem[], newInputs: AgentInputItem[]) => {
const combined = [...history, ...newInputs];
return combined.slice(Math.max(0, combined.length - 6)); // keep last N items
};
const r1 = await agent.run({ input: "My name is Alice. I prefer vegetarian meals." }, { session, sessionInputCallback });
console.log(r1.output);
const r2 = await agent.run({ input: "Suggest dinner ideas." }, { session, sessionInputCallback });
console.log(r2.output);
<Short, action-oriented description>
This ExecPlan is a living document. The sections Progress, Surprises & Discoveries, Decision Log, and Outcomes & Retrospective must be kept up to date as work proceeds.
This ExecPlan must be maintained in accordance with .agent/PLANS.md.
Purpose / Big Picture
Explain, in a few sentences, what someone gains after this change and how they can see it working.
Progress
Use checkboxes and timestamps. Every stopping point must be recorded here, even if it requires splitting a task (“done” vs “remaining”).
- [ ] (YYYY-MM-DD HH:MMZ) …
Surprises & Discoveries
Document unexpected behavior, bugs, performance tradeoffs, or “this assumption was wrong” discoveries. Include short evidence (test output is ideal).
Decision Log
- Decision: …
Rationale: … Date/Author: …
Outcomes & Retrospective
Summarize what was achieved, what remains, and the main lessons learned.
Context and Orientation
Assume the reader knows nothing about this repo. Define any non-obvious term you use and point to the concrete file(s)/command(s) where it appears.
Include (as relevant):
- key repo paths and what they are for
- exact commands to run (include working directory assumptions)
- constraints (permissions, sandbox, network)
- IDs needed to resume work (e.g.
threadId,SESSION_ID, JSONL file paths, SQLite DB path)
Plan of Work
Describe the sequence of edits and additions in prose. Name the exact files and functions/modules involved.
Concrete Steps
List the exact commands to run and what outputs should look like.
Validation and Acceptance
Describe how to prove the behavior works (tests, CLI invocations, or other observable checks). Phrase acceptance as behavior with specific inputs/outputs.
Idempotence and Recovery
Describe how steps can be safely re-run and how to recover from partial failures.
Artifacts and Notes
Include short, focused transcripts and evidence that prove progress and correctness.
Interfaces and Dependencies
Be prescriptive: name the libraries, modules, and interfaces that must exist when the plan is complete.
# Example execpolicy rules (prefix_rule language; Starlark syntax).
#
# Use as a starting point. Apply the principle of least privilege:
# - allow safe, read-only inspection commands
# - prompt for potentially destructive operations
# - forbid privilege escalation and obvious footguns
prefix_rule(
pattern = ["git", "status"],
decision = "allow",
justification = "Safe repository status check.",
match = [["git", "status"], "git status"],
)
prefix_rule(
pattern = ["git", "diff"],
decision = "allow",
justification = "Safe diff inspection.",
match = [["git", "diff"], "git diff --name-only"],
)
prefix_rule(
pattern = [["rg", "ripgrep"]],
decision = "allow",
justification = "Fast repository search.",
match = ["rg \"TODO\" ."],
)
prefix_rule(
pattern = ["git", "push"],
decision = "prompt",
justification = "Remote writes are sensitive; require explicit approval.",
match = [["git", "push"], "git push origin main"],
)
prefix_rule(
pattern = ["sudo"],
decision = "forbidden",
justification = "Never escalate privileges in agentic workflows.",
match = [["sudo", "true"], "sudo -n true"],
)
prefix_rule(
pattern = ["rm", "-rf"],
decision = "forbidden",
justification = "Destructive recursive delete is too risky; use targeted deletions with review.",
match = [["rm", "-rf", "/tmp/something"]],
)
Read the repository, run the test suite, identify the minimal change needed to make all tests pass, implement only that change, and stop. Do not refactor unrelated files.
Constraints:
- If a change is risky, explain the risk and propose the safest alternative.
- Re-run the failing tests after the fix.
- Summarize what you changed and why.
You are acting as a reviewer for a proposed code change made by another engineer. Focus on issues that impact correctness, performance, security, maintainability, or developer experience. Flag only actionable issues introduced by the change. When you flag an issue, cite the affected file and an exact line range. Prioritize severe issues and avoid nit-level comments unless they block understanding of the diff.
After listing findings, produce an overall correctness verdict ("patch is correct" or "patch is incorrect") with a concise justification and a confidence score between 0 and 1.
Use available tools to ensure file citations and line numbers are correct.
PRAGMA foreign_keys = ON;
-- Optional extension schema for SQLite-first “memory” / RAG.
-- This is intentionally minimal and provider-agnostic.
--
-- Suggested usage:
-- - Store source documents (ADRs, design notes, docs) in `documents`
-- - Store chunked text in `chunks`
-- - Store structured run-scoped notes in `notes`
--
-- If you need vector search, you can add an embedding column (JSON or BLOB) to `chunks`
-- and implement retrieval in application code, or move to a dedicated vector store.
CREATE TABLE IF NOT EXISTS documents (
id TEXT PRIMARY KEY,
source TEXT,
uri TEXT,
title TEXT,
text TEXT NOT NULL,
updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS chunks (
id TEXT PRIMARY KEY,
document_id TEXT NOT NULL REFERENCES documents(id) ON DELETE CASCADE,
chunk_index INTEGER NOT NULL,
text TEXT NOT NULL,
token_count INTEGER,
updated_at TEXT NOT NULL,
UNIQUE(document_id, chunk_index)
);
CREATE TABLE IF NOT EXISTS notes (
id TEXT PRIMARY KEY,
run_id TEXT,
scope TEXT,
key TEXT NOT NULL,
value_json TEXT NOT NULL,
created_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_chunks_document_id ON chunks(document_id);
CREATE INDEX IF NOT EXISTS idx_notes_run_id ON notes(run_id);
CREATE INDEX IF NOT EXISTS idx_notes_scope_key ON notes(scope, key);
PRAGMA foreign_keys = ON;
CREATE TABLE IF NOT EXISTS runs (
id TEXT PRIMARY KEY,
created_at TEXT NOT NULL,
label TEXT,
repo_root TEXT,
metadata_json TEXT
);
CREATE TABLE IF NOT EXISTS events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
run_id TEXT NOT NULL REFERENCES runs(id) ON DELETE CASCADE,
ingested_at TEXT NOT NULL,
event_type TEXT NOT NULL,
thread_id TEXT,
item_id TEXT,
item_type TEXT,
payload_json TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_events_run_id ON events(run_id);
CREATE INDEX IF NOT EXISTS idx_events_thread_id ON events(thread_id);
CREATE INDEX IF NOT EXISTS idx_events_event_type ON events(event_type);
CREATE INDEX IF NOT EXISTS idx_events_item_type ON events(item_type);
Agents SDK + Codex MCP: consistent workflows (single + multi-agent)
This reference distills the “consistent workflows” pattern:
- run Codex CLI as an MCP server so it becomes a tool,
- orchestrate multi-step work using the OpenAI Agents SDK,
- enforce gated handoffs so the workflow is deterministic and auditable,
- enable tracing so you can audit prompts, tool calls, file writes, and timing.
Pair this with an ExecPlan (references/execplans.md) for long work so you can resume after compaction/restarts.
Core components
1. Codex MCP server
- Launch
codex mcp-serveras a long-running stdio MCP process. - Tools exposed:
codex(start) andcodex-reply(continue). - Treat each tool call as a unit of work and capture
threadIdfor resume.
2. Orchestrator agent (Project Manager)
- Owns the plan, gates, and state.
- Schedules specialized worker agents via handoffs.
- Blocks advancement until required artifacts exist and checks pass.
3. Specialized worker agents
- Narrow scopes (designer, frontend, backend, tester, security reviewer).
- Produce specific deliverables only (files or structured reports).
- Hand control back to the orchestrator after delivering.
4. Tracing
- Use traces to debug and audit: prompts, tool calls, handoffs, and timing.
- Group related work into a single trace per workflow run (use a stable
group_idper user/session/run).
Recommended shape: gated handoffs
Use gates as explicit “contracts” between roles:
- File existence: expected artifacts are present at known paths.
- Commands: required checks were run (tests/lint/build) and succeeded.
- Structured boundaries: when a worker returns structured JSON, validate it (schema).
Example gates (typical):
- PM writes
REQUIREMENTS.md,TEST.md,AGENT_TASKS.md→ gate: all exist. - Designer writes
design/design_spec.md→ gate: file exists. - Frontend writes
frontend/index.htmland backend writesbackend/server.js→ gate: both exist. - Tester produces
tests/TEST_PLAN.md(and optionally a script) → gate: exists + runs.
Codex MCP tool parameters (high signal)
In worker instructions, be explicit about safety and scope:
- For analysis-only work:
{ "approval-policy": "never", "sandbox": "read-only" } - For controlled file writes:
{ "approval-policy": "never", "sandbox": "workspace-write" }
Always pass a clear cwd and prefer writing outputs into dedicated subfolders per role.
Make it resumable
Record resume pointers in the ExecPlan:
threadIdvalues (for Codex MCPcodex-reply)- JSONL log paths (if you capture
codex exec --jsonelsewhere) - artifact paths for each stage
- the exact next command/prompt to run
For multi-agent runs, also record:
- orchestrator run identifier
- trace/group identifiers (so you can navigate traces later)
Failure handling
When a gate fails:
- do not “continue anyway”
- ask the owning worker to re-run their scoped work with the missing artifact/check as the only goal
- update the ExecPlan Progress with what failed and what will be retried
Codex CLI – codex exec for automation
Use codex exec when you need:
- a CI/cron-friendly interface
- JSONL progress (
--json) - structured final output (
--output-schema <file>) - resumable multi-stage pipelines (
codex exec resume ...)
Key behavior
- Default output: progress to
stderr, final message only tostdout. --json:stdoutbecomes a JSONL stream (one JSON object per line).
Common flags (high signal)
--json--output-schema <path>--output-last-message <path>/-o <path>--sandbox read-only|workspace-write|danger-full-access--ask-for-approval untrusted|on-failure|on-request|never--full-auto(shortcut; still avoid in untrusted contexts)--cd <dir>--add-dir <dir>(prefer overdanger-full-access)
Resuming
codex exec resume --last "<prompt>"codex exec resume <SESSION_ID> "<prompt>"
Durable logs
Always capture JSONL for audits and debugging:
codex exec --json "<prompt>" | tee codex.jsonlFor multi-hour work, also record the prompt, JSONL path, and resume IDs in an ExecPlan (references/execplans.md).
Then ingest to SQLite:
python3 scripts/codex_jsonl_to_sqlite.py --db codex.sqlite --init
cat codex.jsonl | python3 scripts/codex_jsonl_to_sqlite.py --db codex.sqlite --run-label "my-run"Codex configuration knobs (high signal)
Codex CLI defaults come from ~/.codex/config.toml and can be overridden per run via CLI flags (--config, --profile, etc.).
Approvals and sandbox
approval_policy:untrusted | on-failure | on-request | neversandbox_mode:read-only | workspace-write | danger-full-accesssandbox_workspace_write.network_access: allow outbound network insideworkspace-write- Prefer adding writable paths via
--add-dir/sandbox_workspace_write.writable_rootsinstead ofdanger-full-access.
Exec tooling
features.unified_exec: enable the PTY-backed unified exec tool (beta)features.exec_policy: enforce execpolicy checks for shell/unified exec (experimental; on by default)
Web search
features.web_search_request: allow the model to issue web searches (stable)- CLI:
--search(enables web search tool access)
Project guidance discovery
project_doc_fallback_filenames: additional instruction filenames to treat likeAGENTS.mdproject_doc_max_bytes: cap for combined instructions chain
Codex SDK (TypeScript) – patterns that hold up in production
The Codex SDK (@openai/codex-sdk) wraps the bundled codex binary and communicates via JSONL events over stdin/stdout.
Core concepts
Codex: creates and resumes threadsThread: a conversation that can span multiple turnsthread.run(...): returns a completed turn (buffered)thread.runStreamed(...): streamsThreadEventrecords as the agent works
Defaults (recommendation)
Use the least privileges needed:
- analysis-only:
sandboxMode: "read-only",approvalPolicy: "never" - controlled edits:
sandboxMode: "workspace-write",approvalPolicy: "on-request"or"on-failure"
Structured output (robust boundary)
Treat every “downstream decision” as a structured output boundary:
1. define a Zod schema (strict object, additionalProperties: false behavior) 2. generate JSON Schema (z.toJSONSchema(schema)) 3. pass it as outputSchema 4. JSON.parse(turn.finalResponse) then validate with Zod
Streaming event handling (what matters)
Handle at minimum:
turn.failedand top-levelerrorturn.completed(token usage)item.*lifecycle events- item types:
agent_message(final response)command_execution(command, output, exit code, status)file_change(patch applied, paths)mcp_tool_call(server/tool, args, results/errors)todo_list(plan tool updates)
Resuming threads
Persist the thread.id (after thread.started) and resume with codex.resumeThread(threadId).
If you need end-to-end durability, store thread IDs and JSONL event logs in SQLite (see state-memory-sqlite.md).
Context personalization (Agents SDK): state + memory notes
This reference summarizes a practical “context engineering” approach for personalization using the OpenAI Agents SDK:
- keep a structured state object outside the model (your source of truth),
- distill candidate memories during a run into session notes,
- consolidate session notes into long-term notes at the end (dedupe + conflict rules),
- inject only the relevant slice of state into the model at the start of each run.
Use it when you want an agent to feel consistent across sessions without turning the conversation transcript into your only memory store.
Two kinds of state
1. Session state (short-lived)
- Captures “useful for this run” context.
- Can be re-injected when you trim/summarize history.
- Should be small and aggressively curated.
2. Long-term memory (durable notes + profile)
- Stable preferences and constraints (diet, locale, tone, accessibility needs).
- Things that remain true across sessions, with clear recency rules.
- Stored in your DB (SQLite is usually enough).
The lifecycle (distill → consolidate → inject)
1. Inject (start of run)
- Build system instructions that include:
- a structured profile block (small, stable fields)
- a short memory note list (top-K by relevance + recency)
- Precedence rule (recommended): current user input > session context > long-term memory.
2. Distill (during run)
- Use a dedicated tool (e.g.,
save_memory_note) to capture candidate notes. - Only store durable preferences/constraints; avoid transient facts.
- Save to a session-scoped buffer first (so you can reject/curate).
3. Consolidate (end of run)
- Merge session notes into long-term memory with:
- deduplication
- conflict resolution (often “latest wins”)
- optional TTL/forgetting
- Clear the session buffer after commit.
Agents SDK primitives to use
- Python:
RunContextWrapper[T]lets tools and glue code access your state object. It is not passed to the LLM directly. - Python sessions:
SQLiteSessionpersists conversation history across runs. - TypeScript sessions:
OpenAIConversationsSession(durable) orMemorySession(local dev);sessionInputCallbackfor deterministic trimming/merging.
Guardrails (treat memory as an attack surface)
Memories are effectively “instructions-adjacent” once injected, so treat the pipeline as high risk.
Distillation checks (write-time):
- reject sensitive strings (SSNs, payment details, secrets)
- reject instruction-shaped content (“ignore previous instructions…”, “store this policy…”)
- constrain the note schema to allow only approved fields
Consolidation checks (merge-time):
- “no invention”: do not add facts not present in session notes
- resolve conflicts explicitly (document the rule)
- dedupe aggressively
Injection checks (read-time):
- wrap memory in explicit delimiters (e.g.
<memories>…</memories>) - enforce precedence: user intent overrides memory
- keep token budget small (top-K, short notes, summarize old notes)
Suggested note schema (minimal)
Store each memory note as a small structured object:
type:"preference"|"constraint"|"fact"(avoid"instruction")key: stable identifier (e.g."diet","seat_preference")value: short string or small JSONconfidence: 0..1source:"user"|"system"|"inferred"(prefer"user")created_at: ISO timestamp
In SQLite, keep the source-of-truth in your DB; only inject what’s needed per run.
Evals and logging (don’t guess)
Track whether personalization helps without harming correctness:
- memory_write_rate (too high usually means noisy capture)
- blocked_write_rate (signals adversarial/sensitive writes)
- conflict_rate (how often user overrides memory)
- time_to_personalization (turns until correct preference is applied)
Use tracing to correlate outcomes with memory injection and tool calls.
ExecPlans (durable planning across sessions and compaction)
An ExecPlan is a versioned, living design+execution document that makes long work resumable even when:
- your conversation gets compacted,
- sessions restart,
- multiple agents are involved,
- tool traces are lost.
In practice: treat the ExecPlan as your durable “state machine on disk”.
Core contract
An ExecPlan must be:
- Self-contained: a novice with only a fresh checkout + the plan can execute it.
- Outcome-first: acceptance is observable behavior, not internal attributes.
- Living: progress, surprises, and decisions stay aligned with reality.
- Safe to re-run: include recovery steps and idempotent commands.
The canonical standard lives in .agent/PLANS.md (inside the repo you’re working in).
Where plans live
execplans/execplan-<short-name>.md
Treat execplans/ as a stable namespace: links, CI references, and long-running work should rely on it.
What to record so you can resume
At minimum, record these “resume pointers” in Context and Orientation (and keep them current):
- repo root path (or “run from repo root”)
- branch + current commit hash
- the exact commands already run and their outcomes
- paths to artifacts:
- JSONL logs from
codex exec --json(useteeso you can replay/ingest) - SQLite DB path if you store events/state
- any generated reports
- identifiers needed to resume:
SESSION_IDforcodex exec resume ...(if applicable)threadIdfor@openai/codex-sdkthread resume
If you don’t know an ID (because you’re mid-run), explicitly say what output/event contains it and how to find it.
ExecPlans in multi-agent workflows
Use this division of responsibility:
- Orchestrator
- owns the ExecPlan and keeps it correct
- sets gates (“only proceed if tests pass”)
- records decisions and scope changes
- maintains the canonical list of artifacts
- Workers
- produce scoped outputs only (patches, reports, or JSON)
- do not invent new milestones or widen scope
- add evidence snippets (test output, diffs) when asked
Using ExecPlans with codex exec
Recommended operational pattern:
1. Put the plan on disk first (execplans/...). 2. Run Codex with JSONL logging:
codex exec --json "<prompt>" | tee artifacts/codex.jsonl
3. Record in the plan:
- the prompt (or a stable reference to it),
- the JSONL path,
- the session/thread identifiers (when known).
4. If you need to resume:
codex exec resume <SESSION_ID> "<prompt>"
5. Ingest JSONL to SQLite for audits and querying (optional):
cat artifacts/codex.jsonl | python3 scripts/codex_jsonl_to_sqlite.py --db codex.sqlite --run-label "..."
Using ExecPlans with @openai/codex-sdk
When running SDK-driven threads, record:
- the
threadId(so you can resume deterministically), - the sandbox/approval policy used,
- the schema used for structured output (if any),
- any file-system side effects you expect (created/modified files).
For long work, store streamed events (JSONL) as artifacts and ingest them into SQLite in the same way as CLI JSONL.
Codex as an MCP server + OpenAI Agents SDK orchestration
This is the most flexible route for “agent teams”:
- run Codex as an MCP server (
codex mcp-server) - orchestrate it from a separate process (OpenAI Agents SDK, your own runner, etc.)
- keep each role scoped and auditable
Why MCP matters
MCP turns Codex into a tool your orchestrator can call:
codex: start a session (returnsthreadId)codex-reply: continue a session bythreadId
This separation lets you:
- maintain strict orchestration logic outside Codex
- run multiple Codex sessions in parallel
- record traces and enforce gates
Tool contract (Codex MCP server)
When you run Codex as an MCP server, it exposes two tools:
codex(start a session)codex-reply(continue a session)
High-signal parameters to pass to codex:
prompt(required)cwd(working directory)sandbox(read-only|workspace-write|danger-full-access)approval-policy(untrusted|on-request|on-failure|never)include-plan-tool(if you want Codex to emit plan/todo updates)model(optional override)
codex-reply requires:
threadId(required)prompt(required)
Notes:
- Some surfaces also accept
conversationIdas a deprecated alias forthreadId. - Prefer reading the
threadIdfrom a tool call result’s structured content, when present.
Recommended multi-agent shape (gated handoffs)
Roles:
- Project Manager (orchestrator): owns the plan, gates, and state
- Designer: produces specs
- Developer(s): implement scoped changes
- Tester/Verifier: runs tests, validates behavior, decides pass/fail
- (optional) Reviewer/Security: structured review + policy checks
Gates:
- “file exists” checks for expected artifacts
- “tests passed” checks for each stage
- “structured output validated” checks for tool boundaries
Make the gates durable
For multi-hour runs, put the gating logic in an ExecPlan (references/execplans.md):
- the PM/orchestrator owns the plan file and updates it as gates pass/fail
- the plan records the exact artifact paths, commands, and expected outputs
- the plan records resume pointers (
threadId, JSONL paths, DB path)
Tracing
If using OpenAI Agents SDK, enable tracing and group multi-step workflows into a single trace.
Tooling: dynamic MCP servers
Add specialized MCP servers (filesystem, issue trackers, etc.) and treat each as a bounded capability. Prefer allowlists for enabled tools per server.
Orchestration patterns for agentic coding (battle-tested)
These patterns are designed for correctness-first coding workflows. Use them with:
codex exec(automation + JSONL)@openai/codex-sdk(programmatic control)- Codex MCP server + OpenAI Agents SDK (multi-agent orchestration)
For any work likely to exceed a single session, pair these patterns with an ExecPlan (references/execplans.md) so state survives compaction and restarts.
1) Planner → Executor → Verifier (default)
Planner
- produces a short plan with stop conditions
- produces structured “work spec” (files to touch, commands to run, success criteria)
Executor
- implements the smallest change set
- writes code + updates tests/docs as needed
Verifier
- runs the exact checks (tests/lint/build) and blocks merge if red
- outputs a structured “verdict” object
Key rule: Only the verifier can say “done”.
2) Orchestrator/Worker with gated handoffs (multi-agent)
Use when you have role specialization (design, backend, frontend, QA, security).
Orchestrator responsibilities:
- maintains a single source of truth state:
run_id,threadIds, artifact list, gates - validates gates (file existence, test pass)
- decides which worker runs next and with what context
Worker responsibilities:
- produce specific artifacts only (files, patches, structured report)
- do not self-expand scope
3) Evaluator–optimizer loop (hard problems)
Use when changes are subtle or regressions are likely.
- Optimizer proposes a fix.
- Evaluator checks against explicit rubric and rejects/accepts.
Keep evaluator strict; require citations, line ranges, and reproducible commands.
4) Parallelization (safe parallel work)
Parallelize only when tasks are write-disjoint:
- different directories or services
- no shared config files
- no shared migrations
Otherwise, parallelize analysis only, then merge plans into a single executor.
5) Idempotent, resumable runs
To make runs resumable:
- persist
threadIdand run metadata in SQLite - record JSONL event stream for every run
- make every step safe to re-run (use
--dry-run,git diff,git apply --check)
7) ExecPlan as the “durable state machine”
For long tasks, treat the ExecPlan as the durable source of truth:
- the planner writes/updates the plan (milestones, gates, acceptance)
- the executor implements only what the plan currently says
- the verifier updates Progress with evidence and blocks scope creep
Record “resume pointers” (IDs, artifacts, next command to run) so you can restart from disk with minimal context.
6) Context control (avoid token blowups)
- do not paste entire repos into prompts
- feed targeted diffs and file lists
- use RAG: store “index artifacts” in SQLite (file inventory, module graph) and retrieve on demand
RAG + shared memory (SQLite-first)
This is a practical way to add “memory” shared between runs and agents without introducing infra.
What to use RAG for in coding agents
- recalling prior decisions, constraints, and architecture notes
- retrieving module summaries and API surfaces
- avoiding repeated expensive scans (dependency graphs, file inventories)
SQLite schema (minimal)
Use a simple retrieval store:
documents(id, source, uri, title, text, updated_at)chunks(id, document_id, chunk_index, text, token_count)notes(id, run_id, scope, key, value_json, created_at)for structured memory
Template SQL (copy into your project as needed):
assets/templates/sqlite/rag-schema.sql
If you want vector search:
- store embeddings in
chunks(embedding_json)and do approximate search in-app - or use an external vector DB (only if scale requires it)
Retrieval pattern
1. Ingest/update documents (docs, ADRs, key code entry points). 2. On each agent step, retrieve top-K relevant chunks for the current task. 3. Provide retrieved chunks as context (not instructions) and keep prompts explicit.
Guardrails for memory
- Treat retrieved content as untrusted input (prompt injection risk).
- Keep a strict system/developer instruction layer that cannot be overridden by retrieved text.
- Require structured outputs for decisions that affect code.
Relationship to personalization
RAG is a good fit for retrieving documents and past decisions.
If you want durable user personalization (preferences/constraints), prefer a small structured state + memory notes pipeline (references/context-personalization.md) and inject only the relevant slices into the model each run.
Safety, sandboxing, approvals, and execpolicy
Threat model (practical)
Agentic coding systems are vulnerable to:
- prompt injection (from issues, diffs, docs, screenshots)
- secret exfiltration (logs, environment, files, network)
- destructive commands (rm, curl|bash, privilege escalation)
Default stance
- Run analysis-only by default:
read-onlysandbox,approval-policy: never. - Grant write access only when needed; prefer
workspace-writeoverdanger-full-access. - Avoid “YOLO” flags unless externally sandboxed.
Execpolicy rules (command allow/block)
Use execpolicy to prevent risky commands even when the model requests them.
Typical policies:
- forbid
sudo,curl | sh, package installs in prod, credential tools - prompt for
git push,rm -rf, changing auth/config - allow safe read-only commands (
git status,rg,ls)
Template starter file:
assets/templates/execpolicy/default.rules
Guardrail patterns that work
- require structured outputs for decisions (schemas + validation)
- gate handoffs on artifacts and tests passing
- keep prompts explicit about stop conditions and out-of-scope actions
State, caching, and memory with SQLite
SQLite is the simplest “shared brain” for multi-step and multi-agent runs:
- durable storage of run inputs/outputs
- indexing runs by repo/branch/purpose
- caching expensive computations
- storing
threadIdfor resuming Codex sessions
What to store
At minimum:
runs: one record per workflow run (label, created_at, metadata)events: append-only JSONL events with extracted columns for filtering
Optionally:
artifacts: paths/hashes of important output fileskv_cache: key/value cache for derived summaries
How to ingest JSONL
Use scripts/codex_jsonl_to_sqlite.py:
- safe for repeated ingestion
- stores raw JSON for audit
- extracts
type,thread_id, item types, and usage where available
Use cases
- Resume: retrieve the latest
threadIdby run label, then callcodex-reply. - Audit: answer “what commands did the agent run?” by filtering
command_executionitems. - Cost control: compute total token usage per run (
turn.completedevents). - Caching: memoize expensive steps (diff summaries, dependency graphs).
#!/usr/bin/env python3
import argparse
import datetime as dt
import json
import os
import sqlite3
import sys
import uuid
def utc_now_iso() -> str:
return dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds")
def read_lines(path: str):
if path == "-":
for line in sys.stdin:
yield line
return
with open(path, "r", encoding="utf-8") as f:
for line in f:
yield line
def init_db(conn: sqlite3.Connection):
schema_path = os.path.join(
os.path.dirname(__file__), "..", "assets", "templates", "sqlite", "schema.sql"
)
with open(schema_path, "r", encoding="utf-8") as f:
conn.executescript(f.read())
conn.commit()
def ensure_run(conn: sqlite3.Connection, run_id: str, label: str | None, repo_root: str | None):
cur = conn.execute("SELECT 1 FROM runs WHERE id = ?", (run_id,))
if cur.fetchone():
return
conn.execute(
"INSERT INTO runs (id, created_at, label, repo_root, metadata_json) VALUES (?, ?, ?, ?, ?)",
(run_id, utc_now_iso(), label, repo_root, None),
)
conn.commit()
def extract_fields(event: dict):
event_type = str(event.get("type") or "")
thread_id = None
item_id = None
item_type = None
if event_type == "thread.started":
thread_id = event.get("thread_id")
if event_type.startswith("item.") and isinstance(event.get("item"), dict):
item = event["item"]
item_id = item.get("id")
item_type = item.get("type")
# Some item payloads also include thread IDs in nested params in other surfaces; keep raw JSON anyway.
return event_type, thread_id, item_id, item_type
def main():
parser = argparse.ArgumentParser(
description="Ingest Codex JSONL events (from `codex exec --json`) into SQLite for auditability and reuse."
)
parser.add_argument("--db", required=True, help="Path to SQLite DB file.")
parser.add_argument(
"--init",
action="store_true",
help="Initialize schema (safe to run multiple times).",
)
parser.add_argument(
"--input",
default="-",
help="JSONL input file path, or '-' for stdin (default).",
)
parser.add_argument(
"--run-id",
default=None,
help="Run ID (uuid). If omitted, a new run is created.",
)
parser.add_argument(
"--run-label",
default=None,
help="Optional human label for the run (e.g. 'ci-autofix').",
)
parser.add_argument(
"--repo-root",
default=None,
help="Optional repo root path to associate with the run.",
)
args = parser.parse_args()
conn = sqlite3.connect(args.db)
conn.row_factory = sqlite3.Row
if args.init:
init_db(conn)
run_id = args.run_id or str(uuid.uuid4())
ensure_run(conn, run_id=run_id, label=args.run_label, repo_root=args.repo_root)
inserted = 0
for raw_line in read_lines(args.input):
line = raw_line.strip()
if not line:
continue
try:
event = json.loads(line)
except json.JSONDecodeError:
# Keep ingest robust: ignore non-JSON lines (some runners can mix logs).
continue
if not isinstance(event, dict):
continue
event_type, thread_id, item_id, item_type = extract_fields(event)
payload_json = json.dumps(event, separators=(",", ":"), ensure_ascii=False)
conn.execute(
"INSERT INTO events (run_id, ingested_at, event_type, thread_id, item_id, item_type, payload_json) "
"VALUES (?, ?, ?, ?, ?, ?, ?)",
(run_id, utc_now_iso(), event_type, thread_id, item_id, item_type, payload_json),
)
inserted += 1
conn.commit()
sys.stdout.write(json.dumps({"db": args.db, "run_id": run_id, "inserted": inserted}) + "\n")
if __name__ == "__main__":
main()
#!/usr/bin/env python3
import argparse
import json
import sqlite3
import sys
def main():
parser = argparse.ArgumentParser(description="Summarize Codex runs stored by codex_jsonl_to_sqlite.py")
parser.add_argument("--db", required=True, help="Path to SQLite DB file.")
parser.add_argument("--run-id", default=None, help="Run id to summarize.")
parser.add_argument("--latest", action="store_true", help="Summarize the most recent run.")
args = parser.parse_args()
conn = sqlite3.connect(args.db)
conn.row_factory = sqlite3.Row
run_id = args.run_id
if args.latest:
row = conn.execute("SELECT id FROM runs ORDER BY created_at DESC LIMIT 1").fetchone()
if not row:
sys.stderr.write("No runs found.\n")
sys.exit(1)
run_id = row["id"]
if not run_id:
sys.stderr.write("Provide --run-id or --latest.\n")
sys.exit(2)
run = conn.execute("SELECT * FROM runs WHERE id = ?", (run_id,)).fetchone()
if not run:
sys.stderr.write(f"Run not found: {run_id}\n")
sys.exit(1)
counts = conn.execute(
"SELECT event_type, COUNT(*) AS c FROM events WHERE run_id = ? GROUP BY event_type ORDER BY c DESC",
(run_id,),
).fetchall()
thread = conn.execute(
"SELECT thread_id FROM events WHERE run_id = ? AND thread_id IS NOT NULL ORDER BY id ASC LIMIT 1",
(run_id,),
).fetchone()
last_agent_message = conn.execute(
"SELECT payload_json FROM events WHERE run_id = ? AND event_type = 'item.completed' AND item_type = 'agent_message' "
"ORDER BY id DESC LIMIT 1",
(run_id,),
).fetchone()
summary = {
"run": dict(run),
"thread_id": thread["thread_id"] if thread else None,
"event_type_counts": {r["event_type"]: r["c"] for r in counts},
"last_agent_message": json.loads(last_agent_message["payload_json"]) if last_agent_message else None,
}
sys.stdout.write(json.dumps(summary, indent=2) + "\n")
if __name__ == "__main__":
main()
#!/usr/bin/env python3
import argparse
import os
import shutil
def main():
parser = argparse.ArgumentParser(
description="Initialize a minimal .agent workspace (AGENTS.md + PLANS.md) for Codex-driven planning."
)
parser.add_argument(
"--dir",
default=".",
help="Target repository directory (default: current directory).",
)
parser.add_argument(
"--no-execplans",
action="store_true",
help="Do not create execplans/ or copy the ExecPlan template.",
)
args = parser.parse_args()
target_dir = os.path.abspath(args.dir)
agent_dir = os.path.join(target_dir, ".agent")
os.makedirs(agent_dir, exist_ok=True)
skill_root = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
templates_dir = os.path.join(skill_root, "assets", "templates", "agent")
for name in ["AGENTS.md", "PLANS.md"]:
src = os.path.join(templates_dir, name)
dst = os.path.join(agent_dir, name)
if os.path.exists(dst):
continue
shutil.copyfile(src, dst)
if not args.no_execplans:
execplans_dir = os.path.join(target_dir, "execplans")
os.makedirs(execplans_dir, exist_ok=True)
src = os.path.join(skill_root, "assets", "templates", "execplan.md")
dst = os.path.join(execplans_dir, "execplan-template.md")
if not os.path.exists(dst):
shutil.copyfile(src, dst)
print(agent_dir)
if __name__ == "__main__":
main()
#!/usr/bin/env python3
import argparse
import datetime as dt
import os
import re
import sys
def utc_now_stamp() -> str:
return dt.datetime.now(dt.timezone.utc).strftime("%Y-%m-%d %H:%MZ")
def slugify(value: str) -> str:
value = value.strip().lower()
value = re.sub(r"[^a-z0-9]+", "-", value)
value = value.strip("-")
return value or "plan"
def read_template(skill_root: str) -> str:
template_path = os.path.join(skill_root, "assets", "templates", "execplan.md")
with open(template_path, "r", encoding="utf-8") as f:
return f.read()
def main() -> int:
parser = argparse.ArgumentParser(
description="Create a new ExecPlan file under execplans/ from the codex-sdk ExecPlan template."
)
parser.add_argument(
"--dir",
default=".",
help="Target repository directory (default: current directory).",
)
parser.add_argument(
"--name",
required=True,
help="Short name for the plan (used in filename). Example: 'migrate-bun' or 'add-mcp-tools'.",
)
parser.add_argument(
"--title",
default=None,
help="Title for the plan (first heading). Defaults to a title-cased variant of --name.",
)
parser.add_argument(
"--force",
action="store_true",
help="Overwrite if the plan file already exists.",
)
args = parser.parse_args()
target_dir = os.path.abspath(args.dir)
execplans_dir = os.path.join(target_dir, "execplans")
os.makedirs(execplans_dir, exist_ok=True)
plan_slug = slugify(args.name)
plan_path = os.path.join(execplans_dir, f"execplan-{plan_slug}.md")
if os.path.exists(plan_path) and not args.force:
sys.stderr.write(f"Refusing to overwrite existing plan: {plan_path}\n")
return 2
skill_root = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
template = read_template(skill_root)
title = args.title or " ".join(w.capitalize() for w in plan_slug.split("-"))
lines = template.splitlines()
# Replace the first heading with the provided title.
if lines and lines[0].startswith("# "):
lines[0] = f"# {title}"
else:
lines.insert(0, f"# {title}")
# Replace the first Progress timestamp placeholder.
stamp = utc_now_stamp()
updated = []
replaced_progress = False
for line in lines:
if (not replaced_progress) and "## Progress" in line:
updated.append(line)
replaced_progress = True
continue
if replaced_progress and line.strip().startswith("- [ ] (YYYY-"):
updated.append(f"- [ ] ({stamp}) Initial draft created.")
replaced_progress = False # only replace once
continue
updated.append(line)
with open(plan_path, "w", encoding="utf-8") as f:
f.write("\n".join(updated).rstrip() + "\n")
sys.stdout.write(plan_path + "\n")
return 0
if __name__ == "__main__":
raise SystemExit(main())
Related skills
FAQ
When should I use codex exec vs the SDK?
Use codex exec for scriptable one-shots in CI or cron and @openai/codex-sdk for a server-side app controlling Codex programmatically.
How does it handle durable state?
It persists run metadata and JSONL event logs in SQLite for immutable audit logs, run indexing, and deterministic resume via threadId.