A Simulation Brain: Testing Multi-Agent Systems Without Spending Tokens

Building Lucifer: an open-source AI company you can watch · ⭐ GitHub

🚀 Start here: 1. Meet Lucifer · 2. Why I built it · 3. Architecture

📖 Use it: 4. Getting started · 5. Connect an LLM · 6. Office tour

🛠 Build the engine: 7. Org chart in YAML · 8. Real delegation · 9. Sandboxed tools · 10. Simulation brain

🎨 Build the office: 11. 3D office with R3F · 12. WebSocket + zustand · 13. Visualizing workflow

🧭 What's next: 14. Roadmap · 15. Contributing

Developing a multi-agent system against a real LLM is slow and expensive. Every UI tweak needs a run, every run costs tokens, and no two runs behave the same. Lucifer avoids most of that with a simulation brain: a scripted stand-in for the LLM that goes through exactly the same actions as a real agent. This post shows how it works and how the engine's test suite uses it.

A brain is a tiny interface

The whole design hinges on one small protocol:

class Brain(Protocol):
    """An employee's brain: takes a task and acts through the TaskSession until the task
    is completed, failed, or delegated and waiting on reports.

    `work` runs synchronously in a thread provided by the engine.
    """

    kind: Literal["crew", "sim"]

    def work(self, session: TaskSession) -> WorkUsage: ...

The engine doesn't know or care how a brain decides what to do. It hands over a TaskSession (the only door into the company, see Sandboxed agent tools), waits for work() to return, and then settles the task from what happened: completed, failed, delegated or none of these.

So swapping the LLM for a script changes nothing else. Delegation rules, the workspace sandbox, task statuses, the activity log and every WebSocket event are the real thing.

What the simulation brain does

SimBrain follows the same three-way choice a real manager makes:

def work(self, session: TaskSession) -> WorkUsage:
    agent, task = session.agent, session.task

    def step(status: AgentStatus) -> None:
        session.set_status(status)
        if session.cancelled.wait(self.step_s * random.uniform(0.6, 1.4)):
            raise TaskClosedError("Task was cancelled")

    step("thinking")
    subtasks = session.subtasks()

    # 1. Back from delegating: merge the reports' results into a report file.
    if subtasks:
        session.post_update(f"Merging results from {len(subtasks)} delegated tasks")
        step("working")
        lines = "\n".join(f"- **{t.title}** ({t.status}): {t.result or 'no result'}" for t in subtasks)
        path = session.write_file(f"reports/{task.id}-{_slug(task.title)}.md", f"# {task.title}\n\n{lines}\n")
        session.complete(f"Merged {len(subtasks)} work items into {path}.")
        return _fake_usage(task.description)

    # 2. A manager with matching reports: split the work.
    delegates = _pick_delegates(session)
    if delegates and len(session.ancestors()) < MAX_DELEGATION_DEPTH:
        session.post_update(f'Splitting "{task.title}" into {len(delegates)} parts for the team')
        for report in delegates:
            step("working")
            session.delegate(report.id, f"{task.title}: {report.title} part",
                             f"Handle the {report.title} part of: {task.description or task.title}")
        return _fake_usage(task.description)

    # 3. Otherwise: do the work and write a deliverable.
    session.post_update(f"Starting: {task.title}")
    step("working")
    step("thinking")
    path = session.write_file(f"work/{agent.id}/{task.id}-{_slug(task.title)}.md", ...)
    step("working")
    session.complete(f"{agent.name} finished; the deliverable is at {path}.")
    return _fake_usage(task.description)

A few touches make it feel alive in the office:

• Timed steps with jitter. Each step takes step_s × random(0.6, 1.4) seconds (1.5 s by default), and the agent's status flips between thinking and working. The laptop glow, the orbs rising above the head and the work links all animate exactly as they would with a real model.
• Relevant delegation. _pick_delegates scores each report by how many words of four letters or more from their title and persona appear in the task, and picks the top two. "Write a marketing plan" goes to the CMO, "Design the API" goes to the CTO.
• Bounded depth. MAX_DELEGATION_DEPTH = 2 stops the simulation from always walking to the bottom of the org chart.
• Cancellation-aware waits. session.cancelled.wait(...) sleeps and checks for cancellation in one call, so cancelling a task or pausing the company stops the simulation immediately.
• Fake usage. It returns plausible token counts, so the stats in the office aren't empty.

Picking the brain

LUCIFER_BRAIN=auto   # default: crew if an LLM key is configured, otherwise sim
LUCIFER_BRAIN=sim    # force the simulation, even with keys in .env
LUCIFER_BRAIN=crew   # force CrewAI

The office's top bar shows Simulation or the model name, so you always know which one you're watching.

I keep LUCIFER_BRAIN=sim on whenever I work on the office. All the live-work features, including missions, work links, comets and the follow camera, were built and checked against the simulation before a single real token was spent.

Testing the engine without an API key

The engine's test suite never needs a key. There are three techniques.

1. A fast simulation for end-to-end flows

The fixture builds a real engine with a very fast heartbeat:

@pytest.fixture
def company(tmp_path):
    def build(brain: Brain):
        store = make_store(tmp_path / "state.json")
        engine = Engine(store, brain, heartbeat_s=0.02, max_concurrent=9, workspace_dir=tmp_path / "ws")
        return store, engine
    yield build

With SimBrain(step_s=0.002), a whole company run takes milliseconds. Then the test checks properties instead of exact outputs:

def test_ceo_splits_work_down_the_org_chart_then_merges(company):
    store, engine = company(SimBrain(step_s=0.002))
    engine.start()
    # Written in Vietnamese on purpose: the detected language must reach every subtask.
    root = engine.assign(CreateTaskInput(title="Ra mắt app ghi chú", description="Kế hoạch sản phẩm, kỹ thuật và marketing"))

    until(lambda: store.require_task(root.id).status == "done")

    tasks = store.list_tasks()
    assert len(tasks) > 1
    assert all(t.status == "done" for t in tasks)
    for t in tasks:
        if t.parent_id:
            parent = store.require_task(t.parent_id)
            assert store.require_agent(t.assignee_id).manager_id == parent.assignee_id
    assert store.require_task(root.id).artifacts[0].startswith("reports/")
    assert {t.language for t in tasks} == {"Vietnamese"}

One test covers delegation along the org chart, waking parents, merging, the workspace and language inheritance. until() polls with a timeout instead of sleeping for a fixed time, so the test is fast and not flaky.

2. A purpose-built brain for edge cases

Because a brain is just a class with work(), a test can write one in a few lines to hit a specific path:

class _FailingChildBrain:
    kind = "sim"

    def work(self, session: TaskSession) -> WorkUsage:
        if session.agent.id == "lucifer" and not session.subtasks():
            session.delegate("michael", "Subtask", "")
            return WorkUsage()
        if session.agent.id == "michael":
            raise RuntimeError("API is down")
        session.complete("merged")
        return WorkUsage()


def test_brain_error_fails_the_task_and_wakes_the_parent(company):
    store, engine = company(_FailingChildBrain())
    engine.start()
    root = engine.assign(CreateTaskInput(title="Root task"))

    until(lambda: store.require_task(root.id).status == "done")
    [child] = store.subtasks(root.id)
    assert child.status == "failed"
    assert "API is down" in (child.result or "")

This checks that a crashing brain fails its task with the error as the reason, and that the failure still wakes the parent so the company doesn't hang.

3. Real CrewAI against a fake HTTP API

The simulation can't tell you whether the CrewAI integration works. For that, tests/test_crew.py runs the real CrewBrain against an Anthropic API faked at the HTTP layer with httpx.MockTransport:

SCRIPT = [
    _tool_use("tu_1", "write_file", {"path": "api/design.md", "content": "# API design\n"}),
    _tool_use("tu_2", "complete_task", {"summary": "Wrote api/design.md"}),
    _message([{"type": "text", "text": "Done: the API is designed in api/design.md."}], "end_turn"),
]

def handler(request: httpx.Request) -> httpx.Response:
    if request.url.path.endswith("/count_tokens"):  # CrewAI counts tokens before every turn
        return httpx.Response(200, json={"input_tokens": 100})
    requests.append(request)
    return httpx.Response(200, json=SCRIPT[min(len(requests), len(SCRIPT)) - 1])

The test then asserts that the tools really ran (the file exists, the task is done), and that each request body carried output_config.effort and fallbacks, plus the beta header. The same pattern covers the Cloudflare gateway path. It catches the bugs that matter most when a library upgrade changes how CrewAI builds requests, and it still costs nothing.

Takeaways for your own agent systems

1. Put the rules in a session object, not in the brain. Then any brain, real or fake, follows them.
2. Make the brain an interface with one method. Simulations and test doubles become trivial.
3. Simulate timing and status, not just results. The UI needs the rhythm of real work to be developed properly.
4. Test properties, not transcripts. "Every subtask's assignee reports to the parent's assignee" survives any change in wording.
5. Fake the LLM at the HTTP layer for the one test that proves the real integration works.

Run the suite yourself:

cd lucifer-engine && uv run pytest

The tests are in lucifer-engine/tests and the simulation brain is in brains/sim.py. Next, we move to the browser: building the 3D office with React Three Fiber.

---

Lucifer is open source under the MIT license. Clone it, run it for free with the simulation brain, and ⭐ star it if you'd like to follow along: github.com/siri2moon/lucifer.

Next in the series: Building a 3D Office with React Three Fiber: Yacht, Ocean and Angels.