Contributing

git clone https://github.com/collinear-ai/yc-bench.git
cd yc-bench

uv sync

# .env
ANTHROPIC_API_KEY="sk-ant-..."
GEMINI_API_KEY="AIza..."
OPENROUTER_API_KEY="sk-or-v1-..."
OPENAI_API_KEY="sk-..."

uv run yc-bench --help

uv add yc-bench[postgres]
export DATABASE_URL="postgresql://user:pass@localhost/ycbench"

uv run yc-bench run \
  --model gemini/gemini-2.0-flash-exp \
  --seed 1 \
  --config tutorial

# Set DB path for manual testing
export DATABASE_URL="sqlite:///db/test_manual.db"

# Initialize world
uv run yc-bench run --model gemini/gemini-2.0-flash-exp --seed 42 --config easy
# (Ctrl+C after first turn to keep DB)

# Query state manually
uv run yc-bench company status
uv run yc-bench employee list
uv run yc-bench market browse --limit 10
uv run yc-bench task list

from __future__ import annotations  # Always first

import json                        # stdlib
import logging
from dataclasses import dataclass

from sqlalchemy.orm import Session  # third-party
from pydantic import BaseModel

from ..db.models import Company     # local
from ..config import get_config

# cli/task_commands.py
from . import get_db, json_output, error_output

@task_app.command("priority")
def set_task_priority(
    task_id: UUID = typer.Option(..., help="Task ID to update"),
    priority: int = typer.Option(..., help="Priority level (1-10)"),
):
    """Set the priority level for a task."""
    with get_db() as db:
        task = db.query(Task).filter(Task.id == task_id).first()
        if not task:
            error_output("Task not found")
        
        task.priority = priority
        db.flush()
        
        json_output({
            "task_id": str(task_id),
            "priority": priority,
            "status": "updated",
        })

uv run yc-bench task priority --task-id <uuid> --priority 5

def apply_prestige_decay(db: Session, company_id: UUID, days_elapsed: float) -> None:
    wc = get_world_config()
    if wc.prestige_decay_per_day <= 0 or days_elapsed <= 0:
        return
    
    # New formula: exponential decay instead of linear
    decay_rate = Decimal(str(wc.prestige_decay_per_day))
    floor = Decimal(str(wc.prestige_min))
    
    rows = db.query(CompanyPrestige).filter(CompanyPrestige.company_id == company_id).all()
    for row in rows:
        # Exponential: prestige *= (1 - rate)^days
        multiplier = (Decimal(1) - decay_rate) ** Decimal(str(days_elapsed))
        row.prestige_level = max(floor, row.prestige_level * multiplier)
    
    db.flush()

uv run yc-bench run --model gemini/gemini-2.0-flash-exp --seed 1 --config medium

# agent/runtime/my_runtime.py
from __future__ import annotations

from .base import AgentRuntime
from .schemas import RuntimeTurnRequest, RuntimeTurnResult

class MyRuntime(AgentRuntime):
    def __init__(self, settings, command_executor):
        self.settings = settings
        self.command_executor = command_executor
        self.sessions = {}  # session_id -> conversation history
    
    def run_turn(self, request: RuntimeTurnRequest) -> RuntimeTurnResult:
        session_id = request.session_id
        user_input = request.user_input
        
        # 1. Load/create conversation history
        if session_id not in self.sessions:
            self.sessions[session_id] = []
        
        history = self.sessions[session_id]
        history.append({"role": "user", "content": user_input})
        
        # 2. Call your LLM
        response = self._call_llm(history)
        
        # 3. Execute tool calls
        tool_results = []
        for tool_call in response.get("tool_calls", []):
            result = self.command_executor(tool_call["command"])
            tool_results.append(result)
        
        # 4. Check for sim resume
        resume_payload = None
        checkpoint_advanced = False
        for result in tool_results:
            if "sim resume" in result.get("command", ""):
                resume_payload = result.get("payload")
                checkpoint_advanced = True
        
        # 5. Save assistant response
        history.append({"role": "assistant", "content": response["text"]})
        
        return RuntimeTurnResult(
            final_output=response["text"],
            resume_payload=resume_payload,
            checkpoint_advanced=checkpoint_advanced,
            turn_cost_usd=response.get("cost", 0.0),
            raw_result={"tool_calls": tool_results},
        )
    
    def clear_session(self, session_id: str):
        self.sessions.pop(session_id, None)
    
    def _call_llm(self, history):
        # Your LLM API call here
        pass

# agent/runtime/factory.py
from .my_runtime import MyRuntime

def build_runtime(settings, command_executor):
    if settings.model.startswith("my-provider/"):
        return MyRuntime(settings, command_executor)
    # ... existing logic

uv run yc-bench run --model my-provider/my-model --seed 1 --config tutorial

# config/schema.py
class WorldConfig(BaseModel):
    # ... existing fields ...
    
    # New parameter: max simultaneous tasks per employee
    max_tasks_per_employee: int = 3

# cli/task_commands.py
from ..config import get_world_config

@task_app.command("assign")
def assign_task(task_id: UUID, employee_id: UUID):
    with get_db() as db:
        cfg = get_world_config()
        
        # Check constraint
        current_tasks = db.query(TaskAssignment).filter(
            TaskAssignment.employee_id == employee_id
        ).count()
        
        if current_tasks >= cfg.max_tasks_per_employee:
            error_output(f"Employee already assigned to {current_tasks} tasks (max {cfg.max_tasks_per_employee})")
        
        # ... existing logic ...

# config/presets/hard.toml
[world]
max_tasks_per_employee = 2  # Harder: less multitasking allowed

git checkout -b feature/my-new-feature

git add .
git commit -m "Add task priority CLI command"

git push origin feature/my-new-feature