Skip to main content

Documentation Index

Fetch the complete documentation index at: https://mintlify.com/joseluis-dev/harness-ai/llms.txt

Use this file to discover all available pages before exploring further.

Runtime Python is the internal execution engine of Harness AI. It lives exclusively on the private Docker network and is never reachable from the public internet. Its job is to do all the heavy lifting that Harness Core deliberately refuses to do: running agents, orchestrating LangGraph workflows, performing RAG retrieval, generating embeddings, dispatching CLI tools, and persisting audit records. The BFF’s only contact with this service is a fetch() over the private network — the runtime never calls back to the BFF, and the BFF never opens a Postgres or Redis connection of its own.

​
Architecture

The runtime ships as a single container that supervises two processes via entrypoint.sh:
  • uvicorn runtime.api.main:app on port 8000 (the FastAPI HTTP surface)
  • rq worker executions (the in-process RQ execution worker)
Both processes share the same Python environment, the same database connection pool configuration, and the same harness_db PostgreSQL instance. The dual-process design avoids the overhead of a separate worker container while keeping the HTTP surface and the job queue cleanly separated in code.
workers=1 is intentional in the uvicorn invocation. A second uvicorn worker process would compete with the RQ worker for the Redis connection and could double-enqueue jobs. The entrypoint supervises both processes and restarts either on failure.

​
Internal Module Structure

runtime-python/src/
  workers/
    execution_worker.py       ← RQ job: drives execution lifecycle
    document_ingestion_worker.py
    embedding_worker.py
    watcher_worker.py

  agents/
    base_agent.py
    sql_readonly_agent.py
    document_assistant.py
    expediente_assistant.py

  workflows/
    expediente_review_graph.py
    report_generation_graph.py

  rag/
    ingestion.py
    chunking.py
    retrieval.py
    citations.py
    permissions.py

  watchers/
    directory_watcher.py
    event_deduplicator.py
    file_stability.py

  tools/
    tool_gateway.py
    tool_registry.py
    tool_policy.py

  mcp/
    client.py
    transports.py

  cli/
    runner.py
    sandbox.py
    allowed_commands.py

  models/
    gateway_client.py
    routing.py

  audit/
    logger.py
    event_builder.py

  policies/
    evaluator.py
    rules.py

​
API Schemas

The runtime’s FastAPI surface uses Pydantic v2 models with extra="forbid" throughout, ensuring the contract is enforced at the boundary. 
# apps/runtime-python/src/runtime/api/main.py

class ExecutionCreateRequest(BaseModel):
    model_config = ConfigDict(extra="forbid")

    source_type: str = Field(min_length=1, max_length=128)
    intent:      str = Field(min_length=1, max_length=128)
    payload:     dict[str, Any] = Field(default_factory=dict)
    constraints: Optional[dict[str, Any]] = None


class ExecutionCreateResponse(BaseModel):
    id:     uuid.UUID
    status: str


class ExecutionResponse(BaseModel):
    id:           uuid.UUID
    status:       str
    source_type:  str
    intent:       str
    payload:      dict[str, Any]
    created_at:   datetime
    updated_at:   datetime
    completed_at: Optional[datetime] = None

​
The Execution Lifecycle

Every execution travels through four states. Only the runtime may advance or terminal-ise an execution — the BFF never writes to the executions table. 
# apps/runtime-python/src/runtime/db/models.py

class ExecutionStatus(str, enum.Enum):
    QUEUED    = "queued"
    RUNNING   = "running"
    COMPLETED = "completed"
    FAILED    = "failed"
1

QUEUED

POST /internal/executions inserts the row with status=queued and commits the transaction. The enqueue to Redis/RQ happens after the commit so a broker outage cannot roll back the row. The BFF receives 201 { id, status: "queued" }.
2

RUNNING

The RQ worker picks up the job, performs a pre-flight check (execution must still be queued), and transitions to running. An execution.running audit event is emitted.
3

COMPLETED

On successful completion the worker transitions to completed, sets completed_at, and emits an execution.completed audit event.
4

FAILED

Any unhandled exception during work transitions the execution to failed with completed_at set. An execution.failed audit event is emitted with payload.error_code = "INTERNAL_ERROR".

​
Idempotency

The worker is idempotent on the execution UUID. If a broker re-delivers a job whose execution is already in a terminal state (completed or failed), the pre-flight check detects it and returns without writing any rows. 
# apps/runtime-python/src/runtime/workers/execution_worker.py

def process_execution(execution_id_str: str) -> dict[str, Any]:
    execution_id = uuid.UUID(execution_id_str)

    # 1. Pre-flight: bail out if not in QUEUED state.
    initial = asyncio.run(_preflight())
    if initial is None:
        return {"execution_id": str(execution_id), "final_status": "missing"}
    if initial.status != ExecutionStatus.QUEUED:
        return {"execution_id": str(execution_id), "final_status": initial.status.value}

    # 2. Transition: QUEUED → RUNNING
    previous = asyncio.run(_update_status(execution_id, new_status=ExecutionStatus.RUNNING))
    record_transition(execution_id, from_status=previous, to_status=ExecutionStatus.RUNNING,
                      event_type="execution.running")

    # 3. Work, then RUNNING → COMPLETED
    try:
        time.sleep(FAKE_WORK_SECONDS)
        asyncio.run(_update_status(execution_id, new_status=ExecutionStatus.COMPLETED,
                                   set_completed_at=True))
        record_transition(execution_id, from_status=ExecutionStatus.RUNNING,
                          to_status=ExecutionStatus.COMPLETED, event_type="execution.completed")
        return {"execution_id": str(execution_id), "final_status": "completed"}
    except Exception as exc:
        # 4. Failure path: RUNNING → FAILED
        asyncio.run(_update_status(execution_id, new_status=ExecutionStatus.FAILED,
                                   set_completed_at=True))
        record_transition(execution_id, from_status=ExecutionStatus.RUNNING,
                          to_status=ExecutionStatus.FAILED, event_type="execution.failed",
                          payload={"error_code": "INTERNAL_ERROR"})
        return {"execution_id": str(execution_id), "final_status": "failed"}

​
ORM Models

# apps/runtime-python/src/runtime/db/models.py

class Execution(Base):
    __tablename__ = "executions"

    id:           Mapped[uuid.UUID]           # gen_random_uuid() server default
    status:       Mapped[ExecutionStatus]     # PostgreSQL ENUM, NOT NULL
    source_type:  Mapped[str]                 # VARCHAR(128)
    intent:       Mapped[str]                 # VARCHAR(128)
    payload:      Mapped[dict[str, Any]]      # JSONB, default '{}'
    created_at:   Mapped[datetime]
    updated_at:   Mapped[datetime]
    completed_at: Mapped[Optional[datetime]]


class AuditEvent(Base):
    __tablename__ = "audit_events"

    id:           Mapped[uuid.UUID]
    execution_id: Mapped[Optional[uuid.UUID]] # nullable (Phase 2+)
    event_type:   Mapped[str]                 # VARCHAR(128)
    from_status:  Mapped[Optional[str]]
    to_status:    Mapped[Optional[str]]
    actor_type:   Mapped[str]                 # default 'system'
    actor_id:     Mapped[Optional[str]]       # VARCHAR(128)
    payload:      Mapped[Optional[dict]]      # JSONB
    created_at:   Mapped[datetime]

​
The Audit Invariant

The runtime is the sole writer of audit_events. The BFF must never open a direct Postgres connection. No other service — not the model-gateway, not the BFF, not any MCP — inserts into audit_events directly.
This invariant is preserved even for cross-service audit events originating outside the runtime. The model-gateway, for example, emits model.call JSON log lines but never touches Postgres directly. Instead, it forwards those events to the runtime via POST /internal/audit/ingest, which writes the rows through the existing record_transition_async writer.

​
POST /internal/audit/ingest

This Phase 2 endpoint accepts batches of forwarded audit events (today: model.call events from the model-gateway). Each event becomes one audit_events row with execution_id=NULL. 
class AuditIngestRequest(BaseModel):
    model_config = ConfigDict(extra="forbid")

    events: list[dict[str, Any]] = Field(
        ..., min_length=1, max_length=1024,
        description="One or more audit events. Each becomes one audit_events row."
    )


class AuditIngestResponse(BaseModel):
    accepted: int   # number of events successfully written
execution_id=NULL on cross-service rows is intentional. The foreign key constraint is preserved — a non-NULL value still references executions.id — but model calls are decoupled from the RQ execution lifecycle. The Phase 2 migration 0004_audit_optional_execution makes execution_id nullable while keeping the FK in place.

​
Internal API Surface

All endpoints are reachable only on the private Docker network. There is no public Swagger UI (docs_url=None).
MethodPathDescription
GET/healthzLiveness probe. Returns { status, service, version }.
POST/internal/executionsInsert execution row + enqueue RQ job. Returns 201 { id, status }.
GET/internal/executionsList executions, ordered created_at desc. Supports limit (≤200) and offset.
GET/internal/executions/{id}Fetch one execution. Returns 404 for unknown ids.
GET/internal/executions/{id}/eventsFetch audit events for one execution, created_at asc. Returns 200 [] for unknown ids.
POST/internal/audit/ingestAccept forwarded audit events from the model-gateway. 422 on empty events.
For the full endpoint reference, see the Runtime API Reference.

Build docs developers (and LLMs) love

Get started for freeTalk to us