deathforge

Concurrent AI agent orchestration runtime. Written in Crystal. No external async libraries or job queue shards. Fiber-based concurrency with Channel coordination. cgroup v2 sandboxing via raw lib C FFI. SSE endpoint on Crystal's built-in HTTP::Server. Priority dispatch, task pipelines, retry policies, YAML workflows, cancellation, rate limits, structured logging, and graceful shutdown.

Architecture

Scheduler

The scheduler is the central dispatch loop. It holds a registry of Executor instances and an internal priority queue that accepts TaskRequest structs. Higher priority values are dispatched first. A long-running Fiber pulls requests from the queue and spawns worker Fibers up to a configurable concurrency limit.

Each worker Fiber calls the executor's call method, captures the ExecutionResult, and writes it back through a per-task Channel. The scheduler never drops tasks: if concurrency is at capacity, the dispatch loop sleeps and retries until a slot opens.

Two dispatch modes:

• submit: fire the task into the queue and return. The caller reads results through a separate mechanism (SSE broadcast, a shared channel, or a collector).
• submit_sync: block the calling Fiber until the executor returns a result. Useful for synchronous orchestration steps where the agent needs the output before deciding what to do next.

Every completed result is pushed into a ring-buffer ResultStore so the /history and /metrics endpoints can query past executions.

Sandbox

Process isolation uses Crystal's Process module and direct lib C FFI bindings to Linux cgroups v2. The sandbox forks a child process for each tool execution, assigns the child PID to a dedicated cgroup subtree under /sys/fs/cgroup/deathforge, and sets memory and CPU limits before the executor runs.

CgroupSandbox creates a directory per task, writes memory.max, memory.swap.max (set to 0, no swap), and cpu.max (quota/period format or "max" for unlimited). It delegates memory and cpu controllers to child subtrees via cgroup.subtree_control.

The parent process polls with waitpid in non-blocking mode (WNOHANG) plus a monotonic deadline for timeout enforcement. If the deadline passes, it sends SIGKILL to the child and raises SandboxTimeout. It also checks memory.events for oom_kill counters. If the kernel's OOM killer intervened, the sandbox raises SandboxOOM.

After execution (success, timeout, or OOM), the sandbox calls cleanup to kill remaining procs and delete the cgroup directory. No orphaned subtrees.

SSE Layer

The SSE server runs on Crystal's built-in HTTP::Server. Nine routes, no framework:

• /stream: long-lived SSE connection. Pushes result and pipeline_result events as tasks complete. Periodic keepalive comments prevent proxy timeouts.
• /submit: accepts JSON with tool, optional id and priority, plus arbitrary key-value args. Dispatches the task and returns {ok: true, id: ...} immediately.
• /cancel: accepts JSON with id. Marks the task as cancelled. The scheduler checks this flag before and during execution.
• /pipeline: accepts JSON with a steps array. Each step has tool, args, and optional label. Runs the pipeline and streams the aggregate result.
• /workflow: accepts YAML with a stages list. Supports condition (success/failure) and parallel_with for concurrent branches. Template variables like {{label.output}} inject upstream results into downstream args.
• /status: returns active/pending counts, executor list, and cancelled task count.
• /health: returns {status: "alive", version: ...}.
• /metrics: returns result counts, success/failure ratios, average latency, active/pending task counts, and connected SSE client count.
• /history: returns recent results. Accepts ?limit=N&tool=name query params.

Client connections are tracked in a Mutex-protected array. Broadcast iterates the array, writes the SSE wire-format payload, and removes closed connections.

Pipeline

A Pipeline chains multiple tool calls where downstream steps consume upstream output. Each step receives the previous step's output in args["input"]. The pipeline stops on the first failure or runs to completion. PipelineResult carries every step's result in order plus a success flag and total elapsed time.

pipeline = Deathforge::Pipeline.new(scheduler)
pipeline.add_step("shell", {"command" => "echo hello"})
pipeline.add_step("input_echo")
result = pipeline.run

Step two receives "hello\n" as args["input"].

Workflow

A Workflow is a YAML-defined multi-step plan with conditional branching and parallel stages. Template variables like {{label.output}} resolve to prior stage results. Conditions (success, failure) control whether a stage runs. parallel_with groups stages that should run concurrently.

stages:
  - tool: shell
    args: {command: "echo data"}
    label: fetch
  - tool: file_write
    args: {path: "/tmp/out", content: "{{fetch.output}}"}
    condition: success
    label: save

Retry

RetryPolicy defines max retries, backoff strategy (fixed, exponential, linear), and which exit codes are retryable. Retrier wraps submit_sync and applies the policy on failure. Pre-built policies: no_retry, aggressive, conservative.

retrier = Deathforge::Retrier.new(scheduler, Deathforge::RetryPolicy.aggressive)
result = retrier.execute(request)

Rate Limiter

Per-executor sliding window rate limits. set_limit("tool_a", 5, 10) allows 5 calls per 10 seconds. wait_and_allow blocks until a slot opens. Calls without a configured limit always proceed.

Cancellation

CancelRegistry tracks cancelled task IDs. The scheduler checks the registry before dispatch and after execution. Cancelled tasks return an error result with exit code -8. The /cancel endpoint marks a task for cancellation and broadcasts an SSE event.

Logging

Structured JSON line logging on STDERR (or any IO). Each entry carries timestamp, level, message, and metadata. Levels: DEBUG, INFO, WARN, ERROR, FATAL. Set a custom output via Logging.set_output(io, level).

Shutdown

ShutdownHandler traps SIGINT and SIGTERM. On first signal it stops the SSE server, drains the scheduler, then exits. On second signal it force-exits immediately.

Executor Interface

Deathforge::Executor is an abstract class. Subclass it to add a new tool. Required method:

def call(args : Hash(String, String)) : ExecutionResult

Optional overrides:

• timeout_ms: wall-clock limit before the sandbox kills the process. Default 30 seconds.
• memory_limit_bytes: cgroup memory.max value. Default 512 MiB.
• cpu_quota_us: cgroup cpu.max quota in microseconds. Default nil (unlimited).

Built-in executors:

Register a custom executor:

scheduler = Deathforge::Scheduler.new
scheduler.register(MyTool.new)
scheduler.run

Set a rate limit on an executor:

scheduler.rate_limiter.set_limit("shell", max_calls: 10, window_seconds: 60)

Usage

Start the runtime:

require "deathforge"

server = Deathforge.run(port: 8080, concurrency: 8)

This registers all built-in executors, starts the scheduler loop, binds the SSE server, and installs signal handlers for graceful shutdown. The process runs until SIGINT or SIGTERM.

Submit a task:

curl -X POST http://localhost:8080/submit \
  -d '{"tool":"shell","command":"echo hello","id":"t1","priority":"5"}'

Cancel a task:

curl -X POST http://localhost:8080/cancel -d '{"id":"t1"}'

Run a pipeline:

curl -X POST http://localhost:8080/pipeline \
  -d '{"steps":[{"tool":"shell","args":{"command":"echo data"},"label":"gen"},{"tool":"input_echo"}]}'

Run a workflow:

curl -X POST http://localhost:8080/workflow \
  -d 'stages:
  - tool: shell
    args: {command: "echo hello"}
    label: greet
  - tool: input_echo
    condition: success'

Stream results:

curl -N http://localhost:8080/stream

Check metrics:

curl http://localhost:8080/metrics

View history:

curl http://localhost:8080/history?limit=20&tool=shell

Building

make deps    # install shards (currently none required)
make build   # compile release binary
make test    # run specs
make run     # build + run
make format  # auto-format source
make clean   # remove binary and lib/

Project Structure

deathforge/
  shard.yml             shard manifest
  Makefile              build + run tasks
  src/
    deathforge.cr       entry point, top-level API
    deathforge/
      executor.cr       abstract Executor class, ExecutionResult
      scheduler.cr      priority-aware Fiber scheduler with Channel dispatch
      cgroup.cr         lib C FFI + CgroupSandbox helper
      sandbox.cr        fork + cgroup + timeout + OOM detection
      sse_server.cr     SSE HTTP endpoint, nine routes
      builtin.cr        shell, file_read, file_write, http_fetch
      logger.cr         structured JSON line logging
      result_store.cr   ring buffer for execution history
      retry.cr          RetryPolicy, Retrier, backoff strategies
      pipeline.cr       task chaining with output injection
      workflow.cr       YAML-based multi-step workflow DSL
      shutdown.cr       SIGINT/SIGTERM handler, graceful drain
      rate_limiter.cr   sliding window per-executor rate limits
      cancel.cr         CancelRegistry for task cancellation
  spec/
    spec_helper.cr      stub executors for test isolation
    scheduler_spec.cr   scheduler dispatch, concurrency, priority
    sandbox_spec.cr     cgroup setup, cleanup, and error specs
    result_store_spec.cr ring buffer storage and retrieval
    retry_spec.cr       backoff strategies, retry policies
    pipeline_spec.cr    chaining, output injection, failure stop
    features_spec.cr    cancellation, rate limits, logging, shutdown

Requirements

• Crystal >= 1.14.0
• Linux with cgroups v2 mounted at /sys/fs/cgroup
• Root or cgroup write access for sandbox mode (non-sandboxed mode works without privileges)

License

MIT