prefactor_core.managers.agent_instance module

prefactor_core.managers.agent_instance module

Agent instance handle and manager for convenient span creation.

The AgentInstanceManager handles agent instance lifecycle operations, while AgentInstanceHandle provides a high-level interface for managing an agent instance and creating spans within it.

class prefactor_core.managers.agent_instance.AgentInstanceHandle(instance_id: str, client: PrefactorCoreClient)

Bases: object

Handle to an agent instance with convenience methods.

This class provides a clean interface for:

• Starting and finishing the instance
• Creating spans within the instance
• Managing the instance lifecycle

Example

async with client.create_agent_instance(…) as instance: : await instance.start()
async with instance.span(“agent:llm”) as span: : span.set_payload({“model”: “gpt-4”}) # … do work …
await instance.finish()

async create_span(schema_name: str, parent_span_id: str | None = None, payload: dict[str, Any] | None = None, started_at: datetime | None = None) → str

Create a span within this instance and return its ID.

The span stays open until finish_span() is called.

• Parameters:
  • schema_name – Name of the schema for this span.
  • parent_span_id – Optional explicit parent span ID.
  • payload – Optional initial payload (params/inputs) stored on creation.
  • started_at – Optional ISO 8601 start time (defaults to current time).
• Returns: The span ID.

async finish(status: FinishStatus = ‘complete’, timestamp: datetime | None = None) → None

Mark the instance as finished.

Resets the termination monitor (fence + new event) before enqueueing the HTTP finish so stale span responses from the dying run cannot trigger termination on the next run.

• Parameters:
  • status – Terminal status for the instance — one of "complete", "failed", or "cancelled". Defaults to "complete".
  • timestamp – Optional ISO 8601 finish time (defaults to current time).

async finish_span(span_id: str, result_payload: dict[str, Any] | None = None, timestamp: datetime | None = None) → None

Finish a previously created span.

• Parameters:
  • span_id – The ID of the span to finish.
  • result_payload – Optional result data to store on the span.
  • timestamp – Optional ISO 8601 finish time (defaults to current time).

property id : str

Get the instance ID.

• Returns: The unique identifier for this agent instance.

async record_quality(name: str, payload: dict[str, Any] | None = None) → None

Record a quality payload on the instance.

This queues a record_quality operation for the instance.

• Parameters:
  • name – Quality schema name (key in the agent schema version quality_schemas).
  • payload – Quality payload for this name (None to remove).

span(schema_name: str, parent_span_id: str | None = None, payload: dict[str, Any] | None = None)

Create a span within this instance.

This is a convenience method that delegates to the client.

• Parameters:
  • schema_name – Name of the schema for this span.
  • parent_span_id – Optional explicit parent span ID.
  • payload – Optional initial payload (params/inputs) stored on creation.
• Yields: SpanContext for the created span.

async start(timestamp: datetime | None = None) → None

Mark the instance as started.

This queues a start operation for the instance.

• Parameters: timestamp – Optional ISO 8601 start time (defaults to current time).

class prefactor_core.managers.agent_instance.AgentInstanceManager(http_client: PrefactorHttpClient, enqueue: Callable[[Operation], Awaitable[None]])

Bases: object

Manages agent instance lifecycle operations.

This class provides a high-level interface for agent instance operations. Registration is done synchronously to get the API-generated ID, while start/finish operations are queued for async processing.

Example

manager = AgentInstanceManager(http_client, enqueue_func)

Register a new instance (synchronous - returns API-generated ID)

instance_id = await manager.register(

agent_id=”my-agent”, agent_version={“name”: “1.0.0”}, agent_schema_version={“version”: “1.0.0”}

)

Start the instance (queued)

await manager.start(instance_id)

Finish the instance (queued)

await manager.finish(instance_id)

async finish(instance_id: str, status: FinishStatus = ‘complete’, timestamp: datetime | None = None) → None

Mark an instance as finished.

Queues a finish operation for the instance.

• Parameters:
  • instance_id – The ID of the instance to finish.
  • status – Terminal status for the instance. Defaults to "complete".
  • timestamp – Optional ISO 8601 finish time (defaults to current time).

async finish_with_idempotency_key(instance_id: str, idempotency_key: str, status: FinishStatus = ‘complete’, timestamp: datetime | None = None) → None

Queue a finish operation using a stable idempotency key.

• Parameters:
  • instance_id – The ID of the instance to finish.
  • idempotency_key – Stable key for idempotent retries.
  • status – Terminal status for the instance. Defaults to "complete".
  • timestamp – Optional ISO 8601 finish time (defaults to current time).

async record_quality(instance_id: str, name: str, payload: dict[str, Any] | None = None) → None

Record a quality payload on an agent instance.

Queues a record_quality operation for the instance.

• Parameters:
  • instance_id – The ID of the instance to update.
  • name – Quality schema name (key in the agent schema version quality_schemas).
  • payload – Quality payload for this name (None to remove).

async register(agent_version: dict[str, Any], agent_schema_version: dict[str, Any], agent_id: str | None = None, instance_id: str | None = None, environment_id: str | None = None, purpose: InstancePurpose | None = None) → str

Register a new agent instance.

Makes a synchronous API call to register the instance and returns the API-generated ID.

• Parameters:
  • agent_id – Agent ID. Omit when using a deployment-scoped token.
  • agent_version – Version information (name, external_identifier, etc.).
  • agent_schema_version – Schema version information.
  • instance_id – Optional ID to forward to the API as id. When provided, the API uses it as the instance ID; when omitted, the API generates one.
  • environment_id – Optional environment ID. Required when using an account-scoped token; omit when using a deployment-scoped token.
  • purpose – Why this instance ran — "live", "smoke_test", or "eval". Omitted (None) lets the API default to "live".
• Returns: The instance ID (API-generated).

async start(instance_id: str, timestamp: datetime | None = None) → None

Mark an instance as started.

Queues a start operation for the instance.

• Parameters:
  • instance_id – The ID of the instance to start.
  • timestamp – Optional ISO 8601 start time (defaults to current time).

async start_with_idempotency_key(instance_id: str, idempotency_key: str, timestamp: datetime | None = None) → None

Queue a start operation using a stable idempotency key.

• Parameters:
  • instance_id – The ID of the instance to start.
  • idempotency_key – Stable key for idempotent retries.
  • timestamp – Optional ISO 8601 start time (defaults to current time).