PRD v0.5 – Agentic Framework (Implementation-Ready)

Owner: Christopher Henry
Status: Ready for Implementation
Last Updated: 2025-10-25
Design Stance: Keep it very simple. File-backed, Dropbox-synced, one orchestrator.

1. Goals (v0 Scope)

Single CLI to drive and inspect the system
Filesystem message bus (inbox/outbox) with UUID correlation
Orchestrator (singleton): reads inbox → executes commands or creates jobs → writes outbox
Queue Manager (one per compute machine): starts/monitors jobs; moves through queued|running|finished|failed
Worker Harness: validates job → launches Claude Code → captures output → updates job
JSON Schema Validator: one utility used everywhere

Out of Scope for v0:

Slack handler (v1)
Multi-step workflows (v1)
Job priority/urgency (v0.1)
Schema migration tools (v0.1)
Multiple orchestrators
Retry logic (v0.1)
Cross-machine job migration

2. Global Config & Paths

2.1 Environment Variables

AGENT_STORE_ROOT: ~/Dropbox/AgentStore (default)
Can override via environment variable

2.2 Global Config File

Location: <AGENT_STORE_ROOT>/config.yaml

agent_store_root: "/home/chenry/Dropbox/AgentStore"
queues_default: "poplar"

orchestrator:
  polling_interval_seconds: 10
  max_messages_per_loop: 50
  enable_job_completion_handler: true

slack:
  token: null  # Reserved for v1

2.3 User State File

Location: ~/.orchestrator_config.json

{
  "username": "chenry",
  "project": null,
  "prd": null
}

Auto-created by CLI on first login command.

3. Directory Layout

AgentStore/
├── config.yaml
├── messaging/
│   ├── inbox/                    # Messages from users/agents
│   ├── processed_inbox/          # Processed messages
│   ├── outbox/                   # Responses to users
│   └── processed_outbox/         # Sent responses
├── users/
│   └── chenry.json
├── projects/
│   └── ModelSEEDpy/
│       ├── project.json
│       ├── queued_prds/
│       ├── running_prds/
│       └── finished_prds/
├── Queues/
│   ├── poplar/
│   │   ├── queue.json
│   │   ├── log.out
│   │   └── Jobs/
│   │       ├── queued_jobs/
│   │       ├── running_jobs/
│   │       ├── finished_jobs/
│   │       └── failed_jobs/
│   └── oak/
│       └── [same structure]
├── JSONValidation/
│   ├── Schemas/
│   │   ├── user.schema.json
│   │   ├── message.schema.json
│   │   ├── project.schema.json
│   │   ├── prd.schema.json
│   │   ├── queue.schema.json
│   │   └── job.schema.json
│   └── errors.out
└── logs/
    ├── orchestrator.log
    ├── queue-poplar.log
    ├── queue-oak.log
    └── events-2025-10.jsonl

4. JSON Contracts (Canonical Schemas)

All timestamps: UTC ISO 8601 with Z suffix: 2025-10-23T21:59:44Z

4.1 User (`users/<id>.json`)

Schema: user.schema.json

{
  "id": "chenry",
  "slack_username": "chenry",
  "name": "Christopher Henry",
  "email": "chenry@anl.gov",
  "permissions": "admin"
}

Fields:

permissions: "admin" or "user"

4.2 Message (`messaging/inbox/<uuid>.json`, `messaging/outbox/<uuid>.json`)

Schema: message.schema.json

{
  "correlation_id": "550e8400-e29b-41d4-a716-446655440000",
  "source": "user",
  "project": "ModelSEEDpy",
  "queue_time": "2025-10-23T21:59:44Z",
  "processed_time": null,
  "command": "research-code",
  "content": {
    "task": "Review the ModelSEEDpy codebase"
  },
  "status": "queued",
  "error": null
}

Fields:

source: "user" | "orchestrator" | "agent"
status: "queued" | "processing" | "done" | "error"
command: See section 5 for valid commands
content: Dictionary of command arguments

Outbox Response Example:

{
  "correlation_id": "550e8400-e29b-41d4-a716-446655440000",
  "source": "orchestrator",
  "project": "ModelSEEDpy",
  "queue_time": "2025-10-23T21:59:44Z",
  "processed_time": "2025-10-23T22:00:15Z",
  "command": "research-code",
  "content": {
    "job_id": "7b5c9f2e-1234-5678-90ab-cdef12345678",
    "queue": "poplar",
    "job_path": "Queues/poplar/Jobs/queued_jobs/7b5c9f2e-1234-5678-90ab-cdef12345678.json"
  },
  "status": "done",
  "error": null
}

4.3 Project (`projects/<name>/project.json`)

Schema: project.schema.json

{
  "id": "ModelSEEDpy",
  "directory": "/home/chenry/projects/ClaudeProjects/ModelSEEDpy",
  "git_repository": "https://github.com/cshenry/ModelSEEDpy",
  "create_time": "2025-10-23T21:59:44Z",
  "update_time": "2025-10-23T21:59:44Z",
  "description": "Repository with ModelSEEDpy code for building models with the ModelSEED method",
  "owner": "chenry",
  "collaborators": ["fliu", "jplfaria"],
  "code_research_document": "/home/chenry/projects/ClaudeProjects/ModelSEEDpy/MODELSEEDPY_CLAUDE_DOCUMENTATION.md",
  "default_queue": "poplar"
}

New Field:

default_queue: Which queue to use for this project's jobs (required)

4.4 PRD (`projects/<name>/*_prds/<uuid>.json`)

Schema: prd.schema.json

{
  "id": "550e8400-e29b-41d4-a716-446655440000",
  "project": "ModelSEEDpy",
  "title": "Improve FBA performance",
  "version": "1.0",
  "owner": "chenry",
  "status": "queued",
  "created_time": "2025-10-23T21:59:44Z",
  "update_time": "2025-10-23T21:59:44Z",
  "content": {
    "problem_statement": "FBA is too slow for large models",
    "goals": [
      "Improve performance by 10x",
      "Reduce memory footprint"
    ],
    "requirements": [
      "Must maintain accuracy",
      "Must be backward compatible"
    ],
    "acceptance_criteria": [
      "Runs 10x faster on test suite",
      "Passes all existing tests"
    ]
  },
  "jobs": [
    "7b5c9f2e-1234-5678-90ab-cdef12345678",
    "8c6d0a3f-2345-6789-01bc-def123456789"
  ],
  "implementation_report": null
}

Fields:

status: "queued" | "running" | "complete"
content: Structured PRD content (problem, goals, requirements, acceptance criteria)
jobs: Array of job UUIDs associated with this PRD (NEW!)
implementation_report: Populated when status becomes "complete"

Note: Job output is NOT stored in PRD. Use job UUID to retrieve output from job file.

4.5 Queue (`Queues/<queue>/queue.json`)

Schema: queue.schema.json

{
  "config": {
    "id": "poplar",
    "worker_count": 10,
    "claude_cli_path": "claude",
    "rest_interval": 60,
    "claude_default_args": ["-p"],
    "max_job_runtime_seconds": 3600
  },
  "runtime": {
    "process_id": null,
    "active_jobs": 0,
    "last_update": null,
    "queued_jobs": 0
  }
}

Config Fields:

worker_count: Max simultaneous jobs
claude_cli_path: Command to invoke Claude Code (usually "claude")
rest_interval: Seconds to sleep between loops
claude_default_args: Default flags for all jobs (e.g., ["-p"])
max_job_runtime_seconds: Timeout for jobs (NEW!)

Runtime Fields:

process_id: PID of queue manager process (reset on restart)
active_jobs: Current count of running jobs
last_update: Last loop timestamp
queued_jobs: Current count of queued jobs

Important: Queue manager MUST reset all runtime fields on startup (handle stale state).

4.6 Job (`Queues/<queue>/Jobs/.../<uuid>.json`)

Schema: job.schema.json

{
  "config": {
    "project": "ModelSEEDpy",
    "working_directory": "/home/chenry/projects/ClaudeProjects/ModelSEEDpy",
    "queue_time": "2025-10-23T21:59:44Z",
    "jobtype": "claude",
    "claude_command": "code-researcher",
    "claude_extra_args": [],
    "task_description": "Review the ModelSEEDpy codebase and document key components",
    "supporting_data": {
      "modelseedpy_dir": "/home/chenry/projects/ClaudeProjects/ModelSEEDpy"
    },
    "timeout_seconds": 1800
  },
  "runtime": {
    "status": "queued",
    "start_time": null,
    "finish_time": null,
    "process_id": null,
    "llm_process_id": null,
    "error": null,
    "output": null
  }
}

Config Fields:

jobtype: Currently only "claude" supported
claude_command: Must match a command in ~/.claude/commands/
claude_extra_args: Per-job additional flags (e.g., ["--dangerously-skip-permissions"])
task_description: Main prompt for Claude
supporting_data: Key-value pairs provided to Claude as context
timeout_seconds: Max runtime before kill (default from queue config)

Runtime Fields:

status: "queued" | "running" | "finished" | "failed"
process_id: PID of worker harness
llm_process_id: PID of Claude CLI subprocess (if captured)
error: Error message if failed
output: Full stdout/stderr from Claude (up to 10MB, stored inline)

Output Storage:

If output < 10MB: Store inline in runtime.output
If output ≥ 10MB: Truncate to 10MB and note in runtime.error

5. CLI Commands

5.1 Command Categories

Documentation:

help - Show usage

State Management:

login <username> - Set username
set-project <project-id> - Set current project
set-prd <prd-id> - Set current PRD

Data Browsing:

list-users - List all users
list-jobs [--status STATUS] [--project PROJECT] - List jobs
list-projects - List all projects
list-prds [--project PROJECT] - List PRDs
list-tasks - List tasks (requires PRD set) (Future)
list-queues - List all queues

Data Viewing:

view-user <user-id> - View user details
view-job <job-id> - View job details
view-project <project-id> - View project details
view-prd <prd-id> - View PRD details
view-queue <queue-name> - View queue status

Agent Commands (create jobs):

create-prd --project PROJECT - Create new PRD
research-code --project PROJECT - Research codebase
create-tasks --project PROJECT --prd PRD - Generate tasks from PRD
implement-tasks --project PROJECT --prd PRD - Implement tasks
create-environment --project PROJECT - Create venv using venvman

5.2 CLI Global Flags

--format json|text      # Output format (default: text)
--queue QUEUE          # Override default queue
--follow SECONDS       # Wait for response (default: immediate return)
--verbose              # Show debug output

5.3 CLI Behavior

Direct Commands (handled by CLI/Orchestrator):

Execute immediately
Return result synchronously
Commands: help, login, set-*, list-*, view-*

Agent Commands (create jobs):

Write message to inbox
Return immediately (unless --follow specified)
Print: Job queued: <uuid>
If --follow N: Wait up to N seconds for outbox response

Error Handling:

If venvman not installed: Print error, exit 1
If project not found: Print error, exit 1
If queue not found: Print error, exit 1

5.4 Examples

# Setup
orchestrator login chenry
orchestrator set-project ModelSEEDpy

# Browse
orchestrator list-projects
orchestrator view-project ModelSEEDpy
orchestrator list-jobs --status running

# Agent command (async)
orchestrator research-code --project ModelSEEDpy
# Output: Job queued: 7b5c9f2e-1234-5678-90ab-cdef12345678

# Agent command (wait for result)
orchestrator research-code --project ModelSEEDpy --follow 30
# Output: [waits up to 30s for job creation confirmation]

# Create environment
orchestrator create-environment --project ModelSEEDpy
# Error if venvman not installed

6. Orchestrator Design

6.1 Main Loop

class Orchestrator:
    def main_loop(self):
        """Main orchestrator loop"""
        while True:
            # 1. Scan inbox for messages
            messages = self.scan_inbox()
            
            # 2. Process messages one at a time
            for msg in messages[:config.max_messages_per_loop]:
                try:
                    self.process_message(msg)
                except Exception as e:
                    self.handle_error(msg, e)
            
            # 3. Check finished jobs (if enabled)
            if config.enable_job_completion_handler:
                self.handle_finished_jobs()
            
            # 4. Sleep
            time.sleep(config.polling_interval_seconds)
    
    def process_message(self, msg):
        """Process a single message"""
        # Validate
        if not validate_json("message", msg):
            self.write_error_response(msg, "Invalid message format")
            return
        
        # Route
        if msg.command in DIRECT_COMMANDS:
            result = DIRECT_COMMANDS[msg.command](msg)
            self.write_outbox(msg.correlation_id, result)
        
        elif msg.command in AGENT_COMMANDS:
            job = self.create_job(msg)
            self.write_outbox(msg.correlation_id, {
                "status": "done",
                "job_id": job.id,
                "queue": job.queue,
                "job_path": job.path
            })
        
        else:
            self.write_outbox(msg.correlation_id, {
                "status": "error",
                "error": f"Unknown command: {msg.command}"
            })
        
        # Move to processed
        self.move_to_processed_inbox(msg)

6.2 Command Routing

Direct Commands:

Command	Handler	Returns
`help`	`handle_help()`	Help text
`list-users`	`handle_list_users()`	Array of user objects
`list-projects`	`handle_list_projects()`	Array of project objects
`list-prds`	`handle_list_prds()`	Array of PRD objects
`list-jobs`	`handle_list_jobs()`	Array of job objects
`list-queues`	`handle_list_queues()`	Array of queue objects
`view-user`	`handle_view_user()`	User object
`view-project`	`handle_view_project()`	Project object
`view-prd`	`handle_view_prd()`	PRD object
`view-job`	`handle_view_job()`	Job object
`view-queue`	`handle_view_queue()`	Queue object

Agent Commands:

Command	Creates Job	Claude Command	Notes
`create-prd`	Yes	TBD	Creates PRD from user input
`research-code`	Yes	`code-researcher`	Documents codebase
`create-tasks`	Yes	TBD	Generates tasks from PRD
`implement-tasks`	Yes	TBD	Implements tasks
`create-environment`	No	N/A	Calls venvman directly

Note: Claude commands marked "TBD" need to be created/specified.

6.3 Queue Selection Logic

def select_queue(msg, project):
    """Select queue for job"""
    # 1. Explicit queue in message
    if "queue" in msg.content:
        queue_name = msg.content["queue"]
        if queue_exists(queue_name):
            return queue_name
        else:
            raise ValueError(f"Queue not found: {queue_name}")
    
    # 2. Project default queue
    if project.default_queue:
        if queue_exists(project.default_queue):
            return project.default_queue
        else:
            raise ValueError(f"Project queue not found: {project.default_queue}")
    
    # 3. Global default
    queue_name = config.queues_default
    if queue_exists(queue_name):
        return queue_name
    else:
        raise ValueError(f"Default queue not found: {queue_name}")

6.4 Job Creation

def create_job(msg):
    """Create job file from message"""
    # Resolve project
    project = load_project(msg.project)
    
    # Select queue
    queue_name = select_queue(msg, project)
    
    # Build job
    job = {
        "config": {
            "project": project.id,
            "working_directory": project.directory,
            "queue_time": utc_now(),
            "jobtype": "claude",
            "claude_command": AGENT_COMMAND_MAPPING[msg.command],
            "claude_extra_args": msg.content.get("claude_extra_args", []),
            "task_description": build_task_description(msg),
            "supporting_data": build_supporting_data(msg, project),
            "timeout_seconds": msg.content.get("timeout", 1800)
        },
        "runtime": {
            "status": "queued",
            "start_time": None,
            "finish_time": None,
            "process_id": None,
            "llm_process_id": None,
            "error": None,
            "output": None
        }
    }
    
    # Validate
    if not validate_json("job", job):
        raise ValueError("Invalid job structure")
    
    # Write to queue
    job_id = generate_uuid()
    job_path = f"Queues/{queue_name}/Jobs/queued_jobs/{job_id}.json"
    write_json(job_path, job)
    
    return {"id": job_id, "queue": queue_name, "path": job_path}

Command to Claude Command Mapping:

AGENT_COMMAND_MAPPING = {
    "create-prd": "prd-creator",         # TBD: Need to create this
    "research-code": "code-researcher",  # Exists
    "create-tasks": "task-creator",      # TBD: Need to create this
    "implement-tasks": "task-implementer" # TBD: Need to create this
}

Task Description Builder:

def build_task_description(msg):
    """Build plain text prompt for Claude"""
    if msg.command == "research-code":
        return f"Review the {msg.project} codebase and document key components, architecture, and APIs."
    
    elif msg.command == "create-tasks":
        prd = load_prd(msg.content["prd"])
        return f"Create implementation tasks for PRD: {prd.title}\n\n{format_prd(prd)}"
    
    elif msg.command == "implement-tasks":
        # Future
        pass
    
    else:
        return msg.content.get("task", "")

Supporting Data Builder:

def build_supporting_data(msg, project):
    """Build supporting data dict"""
    data = {
        "project_directory": project.directory,
        "project_repository": project.git_repository
    }
    
    if "prd" in msg.content:
        prd = load_prd(msg.content["prd"])
        data["prd"] = prd
    
    # Merge any explicit supporting data
    data.update(msg.content.get("supporting_data", {}))
    
    return data

6.5 Job Completion Handling

def handle_finished_jobs(self):
    """Process completed jobs"""
    for queue_name in list_queues():
        finished_dir = f"Queues/{queue_name}/Jobs/finished_jobs/"
        
        for job_file in glob(f"{finished_dir}/*.json"):
            job = load_json(job_file)
            
            # Handle based on command type
            if job.config.claude_command == "code-researcher":
                self.handle_research_complete(job)
            
            elif job.config.claude_command == "task-creator":
                self.handle_tasks_created(job)
            
            # More handlers as needed
            
            # Archive job
            self.archive_job(job)

def handle_research_complete(self, job):
    """Handle completed code research job"""
    # Update project's code research document
    project = load_project(job.config.project)
    
    # Write output to document file
    with open(project.code_research_document, 'w') as f:
        f.write(job.runtime.output)
    
    # Update project timestamp
    project.update_time = utc_now()
    save_project(project)
    
    # Optional: Notify user
    # (Future: send notification)

def handle_tasks_created(self, job):
    """Handle completed task creation job"""
    # Parse task output
    tasks = parse_task_output(job.runtime.output)
    
    # Load PRD
    prd = load_prd(job.config.supporting_data["prd"]["id"])
    
    # Add job to PRD
    prd.jobs.append(job.id)
    
    # Move PRD to running_prds
    prd.status = "running"
    move_prd(prd, "queued_prds", "running_prds")
    
    # Create task files (future)
    # for task in tasks:
    #     create_task_file(task)
    
    save_prd(prd)

def archive_job(self, job):
    """Move job to archive"""
    # Move from finished_jobs/ to archived/
    # Keep last 90 days in finished_jobs, then auto-archive
    # (Simple implementation: leave in finished_jobs for now)
    pass

6.6 Singleton Enforcement

Problem: Prevent multiple orchestrators running simultaneously.

Solution: Lock file

def start_orchestrator():
    """Start orchestrator with lock file"""
    lock_file = f"{AGENT_STORE_ROOT}/.orchestrator.lock"
    
    # Try to create lock file
    try:
        # Atomic create-if-not-exists
        fd = os.open(lock_file, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
        os.write(fd, str(os.getpid()).encode())
        os.close(fd)
    except FileExistsError:
        # Lock exists, check if process alive
        with open(lock_file) as f:
            pid = int(f.read().strip())
        
        if psutil.pid_exists(pid):
            print(f"ERROR: Orchestrator already running (PID {pid})")
            sys.exit(1)
        else:
            # Stale lock, remove and retry
            os.remove(lock_file)
            return start_orchestrator()
    
    # Register cleanup
    atexit.register(lambda: os.remove(lock_file))
    
    # Run orchestrator
    try:
        orchestrator = Orchestrator()
        orchestrator.main_loop()
    except KeyboardInterrupt:
        print("Orchestrator stopped")
    finally:
        os.remove(lock_file)

7. Queue Manager Design

7.1 Main Loop

class QueueManager:
    def __init__(self, queue_name):
        self.queue_name = queue_name
        self.config = self.load_config()
        self.reset_runtime()
    
    def reset_runtime(self):
        """Reset all runtime state on startup"""
        self.config.runtime.process_id = os.getpid()
        self.config.runtime.active_jobs = 0
        self.config.runtime.last_update = None
        self.config.runtime.queued_jobs = 0
        self.save_config()
    
    def main_loop(self):
        """Main queue manager loop"""
        while True:
            # 1. Check running jobs for crashes/timeouts
            self.check_running_jobs()
            
            # 2. Start new jobs if capacity available
            self.start_new_jobs()
            
            # 3. Update queue.json runtime
            self.update_runtime()
            
            # 4. Log to file
            self.log_status()
            
            # 5. Sleep
            time.sleep(self.config.config.rest_interval)
    
    def check_running_jobs(self):
        """Check running jobs for crashes or timeouts"""
        running_dir = f"Queues/{self.queue_name}/Jobs/running_jobs/"
        
        for job_file in glob(f"{running_dir}/*.json"):
            job = load_json(job_file)
            
            # Check if process still alive
            if not psutil.pid_exists(job.runtime.process_id):
                # Worker died
                job.runtime.status = "failed"
                job.runtime.error = "Worker harness process died"
                job.runtime.finish_time = utc_now()
                save_json(job_file, job)
                move_file(job_file, running_dir, f"Queues/{self.queue_name}/Jobs/failed_jobs/")
                continue
            
            # Check for timeout
            if job.runtime.start_time:
                elapsed = (datetime.utcnow() - parse_time(job.runtime.start_time)).total_seconds()
                timeout = job.config.timeout_seconds or self.config.config.max_job_runtime_seconds
                
                if elapsed > timeout:
                    # Kill the process
                    try:
                        os.kill(job.runtime.process_id, signal.SIGTERM)
                        time.sleep(5)
                        if psutil.pid_exists(job.runtime.process_id):
                            os.kill(job.runtime.process_id, signal.SIGKILL)
                    except:
                        pass
                    
                    # Mark failed
                    job.runtime.status = "failed"
                    job.runtime.error = f"Job timeout after {elapsed}s (limit: {timeout}s)"
                    job.runtime.finish_time = utc_now()
                    save_json(job_file, job)
                    move_file(job_file, running_dir, f"Queues/{self.queue_name}/Jobs/failed_jobs/")
    
    def start_new_jobs(self):
        """Start queued jobs if capacity available"""
        # Count active jobs
        running_dir = f"Queues/{self.queue_name}/Jobs/running_jobs/"
        active_count = len(glob(f"{running_dir}/*.json"))
        
        # Check capacity
        if active_count >= self.config.config.worker_count:
            return
        
        # Get queued jobs (FIFO by modification time)
        queued_dir = f"Queues/{self.queue_name}/Jobs/queued_jobs/"
        queued_files = glob(f"{queued_dir}/*.json")
        queued_files.sort(key=os.path.getmtime)
        
        # Start jobs up to capacity
        slots_available = self.config.config.worker_count - active_count
        
        for job_file in queued_files[:slots_available]:
            job = load_json(job_file)
            
            # Update job status
            job.runtime.status = "running"
            job.runtime.start_time = utc_now()
            
            # Launch worker harness
            worker_pid = self.launch_worker(job_file)
            job.runtime.process_id = worker_pid
            
            # Save and move
            save_json(job_file, job)
            move_file(job_file, queued_dir, running_dir)
    
    def launch_worker(self, job_file):
        """Launch worker harness subprocess"""
        cmd = ["python3", "worker_harness.py", job_file]
        proc = subprocess.Popen(cmd, start_new_session=True)
        return proc.pid
    
    def update_runtime(self):
        """Update queue.json runtime fields"""
        running_dir = f"Queues/{self.queue_name}/Jobs/running_jobs/"
        queued_dir = f"Queues/{self.queue_name}/Jobs/queued_jobs/"
        
        self.config.runtime.active_jobs = len(glob(f"{running_dir}/*.json"))
        self.config.runtime.queued_jobs = len(glob(f"{queued_dir}/*.json"))
        self.config.runtime.last_update = utc_now()
        
        self.save_config()

8. Worker Harness Design

8.1 Job Execution

class WorkerHarness:
    def run_job(self, job_path):
        """Execute a job"""
        # 1. Load and validate
        job = load_json(job_path)
        if not validate_json("job", job):
            self.fail_job(job, job_path, "Invalid job file")
            return
        
        # 2. Validate Claude command exists
        if not self.claude_command_exists(job.config.claude_command):
            self.fail_job(job, job_path, f"Unknown Claude command: {job.config.claude_command}")
            return
        
        # 3. Build prompt file
        prompt_file = self.build_prompt_file(job)
        
        # 4. Build Claude command
        cmd = self.build_claude_command(job, prompt_file)
        
        # 5. Execute Claude
        try:
            output = self.execute_claude(cmd, job.config.working_directory)
            job.runtime.output = output[:10485760]  # 10MB limit
            job.runtime.status = "finished"
        except subprocess.TimeoutExpired:
            job.runtime.status = "failed"
            job.runtime.error = "Claude command timeout"
        except Exception as e:
            job.runtime.status = "failed"
            job.runtime.error = str(e)
        
        # 6. Finalize
        job.runtime.finish_time = utc_now()
        save_json(job_path, job)
        
        # 7. Move to finished or failed
        if job.runtime.status == "finished":
            dest = "finished_jobs"
        else:
            dest = "failed_jobs"
        
        move_file(
            job_path,
            f"Queues/{job.config.queue}/Jobs/running_jobs/",
            f"Queues/{job.config.queue}/Jobs/{dest}/"
        )
    
    def claude_command_exists(self, command):
        """Check if Claude command exists"""
        commands_dir = os.path.expanduser("~/.claude/commands/")
        if not os.path.exists(commands_dir):
            return False
        
        return command in os.listdir(commands_dir)
    
    def build_prompt_file(self, job):
        """Build plain text prompt file"""
        prompt = f"{job.config.task_description}\n\n"
        prompt += "Supporting Data:\n"
        
        for key, value in job.config.supporting_data.items():
            prompt += f"- {key}: {value}\n"
        
        # Write to temp file
        temp_file = f"/tmp/job_{uuid.uuid4()}.txt"
        with open(temp_file, 'w') as f:
            f.write(prompt)
        
        return temp_file
    
    def build_claude_command(self, job, prompt_file):
        """Build Claude CLI command"""
        queue_config = load_queue_config(job.config.queue)
        
        cmd = [queue_config.config.claude_cli_path]
        cmd.extend(queue_config.config.claude_default_args)
        cmd.extend(job.config.claude_extra_args)
        cmd.extend([job.config.claude_command, f"@{prompt_file}"])
        
        return cmd
    
    def execute_claude(self, cmd, working_dir):
        """Execute Claude and capture output"""
        proc = subprocess.Popen(
            cmd,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            cwd=working_dir,
            text=True
        )
        
        # Capture output
        output, _ = proc.communicate()
        
        if proc.returncode != 0:
            raise RuntimeError(f"Claude exited with code {proc.returncode}")
        
        return output
    
    def fail_job(self, job, job_path, error):
        """Mark job as failed"""
        job.runtime.status = "failed"
        job.runtime.error = error
        job.runtime.finish_time = utc_now()
        save_json(job_path, job)
        
        move_file(
            job_path,
            f"Queues/{job.config.queue}/Jobs/running_jobs/",
            f"Queues/{job.config.queue}/Jobs/failed_jobs/"
        )

9. JSON Validation Utility

9.1 Validator Interface

class JSONValidator:
    """Centralized JSON validation"""
    
    def __init__(self):
        self.schemas = self.load_all_schemas()
    
    def load_all_schemas(self):
        """Load all schemas on init"""
        schemas = {}
        schema_dir = f"{AGENT_STORE_ROOT}/JSONValidation/Schemas/"
        
        for schema_file in glob(f"{schema_dir}/*.schema.json"):
            type_name = os.path.basename(schema_file).replace(".schema.json", "")
            with open(schema_file) as f:
                schemas[type_name] = json.load(f)
        
        return schemas
    
    def validate(self, type_name, data):
        """Validate JSON data against schema
        
        Returns:
            (ok: bool, errors: list)
        """
        if type_name not in self.schemas:
            return (False, [f"Unknown schema type: {type_name}"])
        
        schema = self.schemas[type_name]
        
        try:
            jsonschema.validate(data, schema)
            return (True, [])
        except jsonschema.ValidationError as e:
            # Log error
            self.log_error(type_name, data, e)
            return (False, [str(e)])
    
    def log_error(self, type_name, data, error):
        """Log validation error to errors.out"""
        error_entry = {
            "timestamp": utc_now(),
            "type": type_name,
            "id": data.get("id", data.get("correlation_id", "unknown")),
            "error": str(error)
        }
        
        error_file = f"{AGENT_STORE_ROOT}/JSONValidation/errors.out"
        with open(error_file, 'a') as f:
            f.write(json.dumps(error_entry) + "\n")

# Global singleton
validator = JSONValidator()

def validate_json(type_name, data):
    """Convenience function"""
    return validator.validate(type_name, data)

10. Security & Permissions

10.1 User Permissions

admin: All commands
user: All commands except:
- Cannot list/view other users
- Cannot modify other users' projects
- Cannot access other users' PRDs/jobs

10.2 Permission Checks

def check_permission(user, command, target=None):
    """Check if user can execute command"""
    if user.permissions == "admin":
        return True
    
    if command in ["list-users", "view-user"]:
        return False
    
    if target and target.owner != user.id:
        return False
    
    return True

11. Error Handling & Recovery

Scenario 1: Orchestrator Crashes

Detection: Manual (user notices messages not processing)
Recovery: Restart orchestrator
Cleanup: Lock file auto-removed by atexit handler
Impact: Messages remain in inbox, processed after restart

Scenario 2: Queue Manager Crashes

Detection: queue.json shows stale process_id
Recovery: Queue manager resets runtime on startup
Cleanup: Running jobs with dead PIDs marked failed
Impact: Running jobs lost, need manual retry

Scenario 3: Worker Harness Crashes

Detection: Queue manager finds dead process_id in running job
Recovery: Automatic - job marked failed
Impact: Job lost, manual retry needed (v0.1: auto-retry)

Scenario 4: Claude Hangs Forever

Detection: Job exceeds timeout_seconds
Recovery: Queue manager kills process
Impact: Job marked failed with timeout error

Scenario 5: Dropbox Sync Fails

Detection: Manual (files not syncing)
Recovery: Fix Dropbox, restart daemon
Impact: System stalls until sync restored

Scenario 6: Invalid JSON Created

Detection: Validator rejects file
Recovery: Check JSONValidation/errors.out
Impact: Depends on which file - could block system

Scenario 7: Disk Full

Detection: Write errors
Recovery: Free space, restart components
Impact: Jobs may fail with write errors

12. Testing Milestones (Detailed)

M0: Foundation (Days 1-2)

Goal: JSON schemas, validator, basic CLI

Implementation:

Create all 6 JSON schemas (user, message, project, prd, queue, job)
Implement JSONValidator class
Create CLI framework with argparse
Implement: help, login, list-users, view-user

Tests:

# Test 1: Validator accepts valid files
python3 test_validator.py --test valid_schemas

# Test 2: Validator rejects invalid files
python3 test_validator.py --test invalid_schemas

# Test 3: Validator logs errors
cat JSONValidation/errors.out

# Test 4: CLI help works
orchestrator help

# Test 5: CLI login creates config
orchestrator login chenry
cat ~/.orchestrator_config.json

# Test 6: CLI lists users
orchestrator list-users

# Test 7: CLI views user
orchestrator view-user chenry

Success Criteria:

✅ All schemas pass validation tests
✅ Invalid files rejected with clear errors
✅ errors.out contains JSONL entries
✅ CLI commands execute without errors
✅ ~/.orchestrator_config.json created correctly

M1: Orchestrator Message Processing (Days 3-4)

Goal: Orchestrator reads inbox, processes direct commands, writes outbox

Implementation:

Implement Orchestrator class with main_loop
Implement singleton lock file
Implement message scanning and validation
Implement direct command handlers (list/view)
Implement outbox writing
Implement processed_inbox moving

Tests:

# Test 1: Start orchestrator
orchestrator-daemon start
# Verify lock file exists

# Test 2: Cannot start second orchestrator
orchestrator-daemon start
# Should error: "Already running"

# Test 3: Manual message test
cat > messaging/inbox/test-001.json << EOF
{
  "correlation_id": "test-001",
  "source": "user",
  "project": "ModelSEEDpy",
  "queue_time": "2025-10-25T10:00:00Z",
  "processed_time": null,
  "command": "list-projects",
  "content": {},
  "status": "queued",
  "error": null
}
EOF

# Wait 15 seconds
sleep 15

# Test 4: Check outbox
cat messaging/outbox/test-001.json
# Should contain list of projects

# Test 5: Check processed inbox
ls messaging/processed_inbox/test-001.json
# Should exist

# Test 6: Test unknown command
# Create message with command: "invalid-command"
# Verify error in outbox

# Test 7: Stop orchestrator
orchestrator-daemon stop
# Verify lock file removed

Success Criteria:

✅ Orchestrator starts, creates lock file
✅ Cannot start multiple orchestrators
✅ Messages processed from inbox
✅ Responses written to outbox
✅ Processed messages moved
✅ Unknown commands return errors
✅ Lock file cleaned up on exit

M2: Job Creation (Days 5-6)

Goal: Orchestrator creates job files in queues

Implementation:

Implement queue selection logic
Implement job creation from message
Implement agent command routing
Add default_queue to project.json
Create test queue: poplar

Tests:

# Test 1: Create test project with queue
cat > projects/TestProject/project.json << EOF
{
  "id": "TestProject",
  "directory": "/tmp/testproject",
  "git_repository": "https://github.com/test/test",
  "create_time": "2025-10-25T10:00:00Z",
  "update_time": "2025-10-25T10:00:00Z",
  "description": "Test project",
  "owner": "chenry",
  "collaborators": [],
  "code_research_document": "/tmp/testproject/docs.md",
  "default_queue": "poplar"
}
EOF

# Test 2: Create poplar queue
mkdir -p Queues/poplar/Jobs/{queued_jobs,running_jobs,finished_jobs,failed_jobs}
cat > Queues/poplar/queue.json << EOF
{
  "config": {
    "id": "poplar",
    "worker_count": 2,
    "claude_cli_path": "claude",
    "rest_interval": 60,
    "claude_default_args": ["-p"],
    "max_job_runtime_seconds": 3600
  },
  "runtime": {
    "process_id": null,
    "active_jobs": 0,
    "last_update": null,
    "queued_jobs": 0
  }
}
EOF

# Test 3: Send research-code command
orchestrator research-code --project TestProject
# Should print: "Job queued: <uuid>"

# Test 4: Verify job file created
ls Queues/poplar/Jobs/queued_jobs/*.json
cat Queues/poplar/Jobs/queued_jobs/*.json

# Test 5: Verify job validates
python3 -c "
import json
from validator import validate_json
with open('Queues/poplar/Jobs/queued_jobs/<uuid>.json') as f:
    job = json.load(f)
ok, errors = validate_json('job', job)
print('Valid:', ok)
"

# Test 6: Test queue override
orchestrator research-code --project TestProject --queue oak
# Should error: "Queue not found: oak"

# Test 7: Check outbox response
cat messaging/outbox/<uuid>.json
# Should contain job_id, queue, job_path

Success Criteria:

✅ Job file created in correct queue directory
✅ Job file validates against schema
✅ Queue selection uses project default
✅ CLI --queue flag overrides default
✅ Invalid queue returns error
✅ Outbox contains job creation confirmation

M3: Queue Manager Loop (Days 7-9)

Goal: Queue manager moves jobs through states (no execution)

Implementation:

Implement QueueManager class
Implement runtime reset on startup
Implement running job checking (crash detection)
Implement job starting (launch no-op worker)
Implement timeout checking
Implement worker_count enforcement
Add logging

Tests:

# Test 1: Start queue manager
queue-manager --queue poplar
# Verify queue.json.runtime updated

# Test 2: Create test job manually
cat > Queues/poplar/Jobs/queued_jobs/test-job-001.json << EOF
{
  "config": {
    "project": "TestProject",
    "working_directory": "/tmp",
    "queue_time": "2025-10-25T10:00:00Z",
    "jobtype": "noop",
    "claude_command": "noop",
    "claude_extra_args": [],
    "task_description": "Test job",
    "supporting_data": {},
    "timeout_seconds": 60
  },
  "runtime": {
    "status": "queued",
    "start_time": null,
    "finish_time": null,
    "process_id": null,
    "llm_process_id": null,
    "error": null,
    "output": null
  }
}
EOF

# Test 3: Wait for job to move to running
sleep 70
ls Queues/poplar/Jobs/running_jobs/test-job-001.json
# Should exist

# Test 4: Check job updated
cat Queues/poplar/Jobs/running_jobs/test-job-001.json
# Should have: process_id, start_time, status="running"

# Test 5: Kill worker manually
kill <process_id>

# Test 6: Wait for crash detection
sleep 70
ls Queues/poplar/Jobs/failed_jobs/test-job-001.json
# Should exist with error: "Worker harness process died"

# Test 7: Test worker_count limit
# Queue 5 jobs, verify only 2 start (worker_count=2)

# Test 8: Test timeout
# Create job with timeout_seconds=10, let it run
# Verify killed after 10 seconds

Success Criteria:

✅ Queue manager starts, resets runtime
✅ Jobs move queued → running
✅ Dead worker processes detected
✅ Crashed jobs moved to failed
✅ Worker count limit enforced
✅ Timeouts detected and jobs killed
✅ queue.json updated each loop

M4: Worker Harness No-op (Days 10-11)

Goal: Worker harness runs dummy script, captures output

Implementation:

Implement WorkerHarness class
Implement job validation
Implement no-op job execution (sleep + echo)
Implement output capture
Implement job status updates
Implement file moving

Tests:

# Test 1: Create no-op worker script
cat > worker_noop.py << EOF
#!/usr/bin/env python3
import sys, time, json
job_file = sys.argv[1]
with open(job_file) as f:
    job = json.load(f)

# Simulate work
print("Starting no-op job...")
time.sleep(5)
print("Job complete!")

# Update job
job['runtime']['status'] = 'finished'
job['runtime']['output'] = 'No-op job completed successfully'
job['runtime']['finish_time'] = '2025-10-25T10:00:00Z'

with open(job_file, 'w') as f:
    json.dump(job, f, indent=2)
EOF
chmod +x worker_noop.py

# Test 2: Modify queue manager to launch no-op worker
# Change launch_worker() to: python3 worker_noop.py <job_file>

# Test 3: Create job, let it run
# Verify moves to finished_jobs after 5 seconds

# Test 4: Check output
cat Queues/poplar/Jobs/finished_jobs/<uuid>.json
# Should have: status="finished", output="No-op job completed successfully"

Success Criteria:

✅ Worker harness validates job
✅ Worker executes and completes
✅ Output captured in job.runtime.output
✅ Job moved to finished_jobs
✅ Job status and timestamps updated

M5: Claude Code Integration (Days 12-14)

Goal: Run real Claude Code commands

Implementation:

Replace no-op worker with real WorkerHarness
Implement Claude command validation
Implement prompt file building
Implement Claude CLI invocation
Implement output capture (10MB limit)
Test with code-researcher command

Tests:

# Test 1: Verify Claude installed
which claude

# Test 2: Verify code-researcher exists
ls ~/.claude/commands/code-researcher

# Test 3: Create real research job
orchestrator research-code --project ModelSEEDpy --follow 300

# Test 4: Monitor job progress
watch -n 5 'ls Queues/poplar/Jobs/*/*.json | xargs -I {} basename {}'

# Test 5: Wait for completion (may take 5-30 minutes)

# Test 6: Check output
cat Queues/poplar/Jobs/finished_jobs/<uuid>.json | jq '.runtime.output'

# Test 7: Verify output size
cat Queues/poplar/Jobs/finished_jobs/<uuid>.json | jq '.runtime.output' | wc -c
# Should be < 10MB

# Test 8: Test with intentional error
# Create job with invalid claude_command
# Verify fails with "Unknown Claude command" error

# Test 9: Test timeout
# Create job with timeout_seconds=30, give long task
# Verify killed after 30 seconds

Success Criteria:

✅ Claude Code command executes
✅ Output captured correctly
✅ Large outputs truncated to 10MB
✅ Invalid commands caught early
✅ Timeouts work correctly
✅ Failed jobs have clear error messages

M6: End-to-End Workflow (Days 15-16)

Goal: Complete user workflow from CLI to result

Implementation:

Implement orchestrator job completion handling
Implement PRD job tracking
Test multiple concurrent jobs
Performance tuning

Tests:

# Test 1: Full workflow
orchestrator login chenry
orchestrator set-project ModelSEEDpy
orchestrator research-code --follow 600

# Test 2: Check project updated
cat projects/ModelSEEDpy/project.json | jq '.update_time'

# Test 3: Check research document
cat <project.code_research_document>

# Test 4: Multiple concurrent jobs
for i in {1..5}; do
  orchestrator research-code --project TestProject &
done

# Wait for all to complete
wait

# Test 5: Verify all jobs processed
ls Queues/poplar/Jobs/finished_jobs/*.json | wc -l
# Should be 5

# Test 6: Test job failure handling
# Create project with non-existent directory
# Verify job fails gracefully

# Test 7: Performance test
# Queue 20 jobs, measure time to complete

Success Criteria:

✅ Complete CLI → job → result workflow works
✅ Project documents updated correctly
✅ Multiple concurrent jobs handled
✅ Failures handled gracefully
✅ System remains stable under load
✅ All provenance tracked (job UUIDs in PRDs)

13. Configuration Reference

13.1 config.yaml (Full Example)

agent_store_root: "/home/chenry/Dropbox/AgentStore"
queues_default: "poplar"

orchestrator:
  polling_interval_seconds: 10
  max_messages_per_loop: 50
  enable_job_completion_handler: true
  singleton_lock_file: ".orchestrator.lock"

slack:
  token: null
  enabled: false

logging:
  level: "INFO"
  events_file: "logs/events-{YYYY-MM}.jsonl"

13.2 Queue Config (Full Example)

{
  "config": {
    "id": "poplar",
    "worker_count": 10,
    "claude_cli_path": "claude",
    "rest_interval": 60,
    "claude_default_args": ["-p", "--dangerously-skip-permissions"],
    "max_job_runtime_seconds": 3600
  },
  "runtime": {
    "process_id": null,
    "active_jobs": 0,
    "last_update": null,
    "queued_jobs": 0
  }
}

14. Open Questions & Future Work

v0.1 (Next Iteration)

Job priority/urgency
Auto-retry failed jobs
Schema migration tools
Better error recovery
Job cancellation
More Claude commands (task-creator, etc.)

v1.0 (Future)

Slack integration
Multi-step workflows
Task dependencies
Cross-project coordination
Web dashboard
API endpoints

15. Success Criteria for v0

✅ All M0-M6 tests pass
✅ CLI usable for daily work
✅ Orchestrator stable for 24+ hours
✅ Queue managers handle 10+ concurrent jobs
✅ Jobs complete successfully
✅ All JSON schemas validated
✅ Error handling works
✅ Documentation complete

This PRD is now ready for implementation with Claude Code!

Start with M0, test thoroughly, then proceed to M1, etc. Each milestone is independently testable and builds on previous work.

FilesExpand file tree

PRD_Agentic_Framework_v0.5_FINAL.md

Latest commit

History

PRD_Agentic_Framework_v0.5_FINAL.md

File metadata and controls

PRD v0.5 – Agentic Framework (Implementation-Ready)

1. Goals (v0 Scope)

2. Global Config & Paths

2.1 Environment Variables

2.2 Global Config File

2.3 User State File

3. Directory Layout

4. JSON Contracts (Canonical Schemas)

4.1 User (users/<id>.json)

4.2 Message (messaging/inbox/<uuid>.json, messaging/outbox/<uuid>.json)

4.3 Project (projects/<name>/project.json)

4.4 PRD (projects/<name>/*_prds/<uuid>.json)

4.5 Queue (Queues/<queue>/queue.json)

4.6 Job (Queues/<queue>/Jobs/.../<uuid>.json)

5. CLI Commands

5.1 Command Categories

5.2 CLI Global Flags

5.3 CLI Behavior

5.4 Examples

6. Orchestrator Design

6.1 Main Loop

6.2 Command Routing

6.3 Queue Selection Logic

6.4 Job Creation

6.5 Job Completion Handling

6.6 Singleton Enforcement

7. Queue Manager Design

7.1 Main Loop

8. Worker Harness Design

8.1 Job Execution

9. JSON Validation Utility

9.1 Validator Interface

10. Security & Permissions

10.1 User Permissions

10.2 Permission Checks

11. Error Handling & Recovery

Scenario 1: Orchestrator Crashes

Scenario 2: Queue Manager Crashes

Scenario 3: Worker Harness Crashes

Scenario 4: Claude Hangs Forever

Scenario 5: Dropbox Sync Fails

Scenario 6: Invalid JSON Created

Scenario 7: Disk Full

12. Testing Milestones (Detailed)

M0: Foundation (Days 1-2)

M1: Orchestrator Message Processing (Days 3-4)

M2: Job Creation (Days 5-6)

M3: Queue Manager Loop (Days 7-9)

M4: Worker Harness No-op (Days 10-11)

M5: Claude Code Integration (Days 12-14)

M6: End-to-End Workflow (Days 15-16)

13. Configuration Reference

13.1 config.yaml (Full Example)

13.2 Queue Config (Full Example)

14. Open Questions & Future Work

v0.1 (Next Iteration)

v1.0 (Future)

15. Success Criteria for v0

4.1 User (`users/<id>.json`)

4.2 Message (`messaging/inbox/<uuid>.json`, `messaging/outbox/<uuid>.json`)

4.3 Project (`projects/<name>/project.json`)

4.4 PRD (`projects/<name>/*_prds/<uuid>.json`)

4.5 Queue (`Queues/<queue>/queue.json`)

4.6 Job (`Queues/<queue>/Jobs/.../<uuid>.json`)