mineproductivity.agents

Auto-generated API reference for the Agents package, rendered from the package's own docstrings. For the narrative overview, dependency rules, and extension guide, see Packages -> Agents.

agents

mineproductivity.agents -- the AI-agent orchestration layer (design spec 11): a model-independent orchestration layer for autonomous and semi-autonomous work, built directly above optimization. Defines how agent work is governed, gated, executed, delegated, and audited -- Agent/Tool/AgentMemory are interface-only extension points; choosing a reasoning backend is exactly the implementation decision this package excludes (§3.1, §4).

Public API (design spec §7) -- every name stable once implementation begins.

Agent

Bases: ABC

The root of every registrable agent type (design spec §8) -- the direct counterpart of optimization.OptimizationModel one layer down, whose category members (§29) are domain roles rather than algorithmic paradigms.

Statelessness. Every subclass, of every category, is stateless -- no instance attribute is mutated by any _act implementation (§29, §32); statefulness lives entirely in Task (§11), so the same registered Agent type can execute many concurrent Task\ s safely.

Qualify, don't coerce (§30): no _act implementation raises for a legitimately incomplete or ambiguous task -- it returns an AgentResult carrying a warning instead.

AgentContext

AgentContext(*, kpi_results=(), analytics_results=(), decision_results=(), twin_snapshot=None, simulation_results=(), optimization_results=())

Bundles the evidence an Agent may need (design spec §8) -- carried once, at construction; every fact an agent reasons over has a stable, structured home in a lower package, never a re-derivation of its own.

Examples:

>>> context = AgentContext()
>>> context.kpi_results
()
>>> context.twin_snapshot is None
True

ApprovalRequest dataclass

ApprovalRequest(task_id, requested_action, *, status=ApprovalStatus.PENDING, approver=None, resolved_at=None)

Bases: BaseValueObject

One pending, approved, or rejected request for a human to authorize a Task before TaskExecutor (design spec §12) resumes it.

Examples:

>>> request = ApprovalRequest(task_id="TASK-1", requested_action="approve_shutdown")
>>> request.status
<ApprovalStatus.PENDING: 'pending'>
>>> request.approver is None
True

ApprovalStatus

Bases: Enum

The three-state approval lifecycle a Task pauses on when PolicyEngine (design spec §10) requires it -- closed; a new approval policy (who must approve what) is an AgentPolicy concern, not a change here.

AgentAuditEntry dataclass

AgentAuditEntry(recorded_at, result, agent_code, scope)

Bases: BaseValueObject

One append-only record: which agent produced which result, for what scope, when (design spec §21).

Examples:

>>> from datetime import timezone
>>> entry = AgentAuditEntry(
...     recorded_at=datetime(2026, 7, 1, tzinfo=timezone.utc),
...     result=AgentResult(task_id="TASK-1"),
...     agent_code="FLEET.ReassignmentAdvisor", scope={"pit": "north"},
... )
>>> dict(entry.scope)
{'pit': 'north'}

AgentAuditTrail

AgentAuditTrail()

Append-only record of every terminal AgentResult produced by any TaskExecutor run in this process (design spec §21). Serializes via core.serialization exactly as every other value object in this platform does.

Examples:

>>> from datetime import timezone
>>> trail = AgentAuditTrail()
>>> trail.record(AgentAuditEntry(
...     recorded_at=datetime(2026, 7, 1, tzinfo=timezone.utc),
...     result=AgentResult(task_id="TASK-1"),
...     agent_code="FLEET.ReassignmentAdvisor", scope={"pit": "north"},
... ))
>>> len(trail.query())
1
>>> len(trail.query(scope={"pit": "south"}))
0

AgentCapabilitySet dataclass

AgentCapabilitySet(agent_code, permissions)

Bases: BaseValueObject

The full set of Permission\ s a registered Agent type carries -- authored and governed, never inferred from an agent's own code (design spec §9).

Examples:

>>> caps = AgentCapabilitySet(
...     agent_code="FLEET.ReassignmentAdvisor",
...     permissions=(Permission(capability="reassign_truck"),),
... )
>>> caps.permissions[0].capability
'reassign_truck'

Permission dataclass

Permission(capability, *, scope=dict())

Bases: BaseValueObject

One capability a registered Agent is authorized to exercise, scoped the same way digital_twin.Twin.scope already is (design spec §9).

Examples:

>>> Permission(capability="approve_shutdown", scope={"pit": "north"}).capability
'approve_shutdown'

AgentMessage dataclass

AgentMessage(from_agent_code, to_agent_code, task_id, content, sent_at)

Bases: BaseValueObject

A structured message exchanged between two agents, or between an agent and a human supervisor (design spec §18). content is an open Mapping[str, Any], never a closed schema, since a message's shape depends entirely on the sending Agent's own category.

Examples:

>>> from datetime import timezone
>>> message = AgentMessage(
...     from_agent_code="SHIFT_SUPERVISOR.NightShift",
...     to_agent_code="FLEET.ReassignmentAdvisor",
...     task_id="TASK-1", content={"kind": "delegation"},
...     sent_at=datetime(2026, 7, 1, tzinfo=timezone.utc),
... )
>>> message.content["kind"]
'delegation'

DelegationRequest dataclass

DelegationRequest(task_id, from_agent_code, to_agent_code, reason)

Bases: BaseValueObject

One agent's request that another take over (or assist with) a Task, carrying the reason for delegation for the eventual audit trail (design spec §18, §21).

Examples:

>>> DelegationRequest(
...     task_id="TASK-1", from_agent_code="SHIFT_SUPERVISOR.NightShift",
...     to_agent_code="FLEET.ReassignmentAdvisor", reason="fleet expertise required",
... ).reason
'fleet expertise required'

ConversationContext dataclass

ConversationContext(task_id, *, turns=())

Bases: BaseValueObject

An ordered history of ConversationTurn\ s for one Task (design spec §15). Append-only in spirit: a new turn produces a new instance -- the same "represent a state change as a new instance" discipline Task.with_state() (§11) applies one level up.

Examples:

>>> ConversationContext(task_id="TASK-1").turns
()

ConversationTurn dataclass

ConversationTurn(speaker, content, occurred_at)

Bases: BaseValueObject

One exchange in a task's dialogue -- speaker is an open string (an Agent's own code, "human", or a Tool's code), never a closed enum (design spec §15).

Examples:

>>> from datetime import timezone
>>> turn = ConversationTurn(
...     speaker="human", content="Proceed with plan B.",
...     occurred_at=datetime(2026, 7, 1, tzinfo=timezone.utc),
... )
>>> turn.speaker
'human'

AgentExecutionError

AgentExecutionError(message, *, details=None)

Bases: MineProductivityError

TaskExecutor raised for a step that should have been structurally valid -- distinct from a legitimately-denied or legitimately-pending case (design spec §30's 'qualify, don't coerce' rule), which returns an AgentResult carrying a warning instead of raising.

AgentValidationError

AgentValidationError(message, *, details=None)

Bases: ValidationError

An AgentMetadata, Task, or TaskState failed validation (design spec §29, §11) -- e.g. an empty code, a Task with no assigned agent code, or a State with empty attributes.

AgentVersionConflictError

AgentVersionConflictError(message, *, details=None)

Bases: RegistrationError

A plugin attempted to re-register an existing Agent or Tool type code with materially different metadata without a version bump, mirroring optimization.OptimizationVersionConflictError (spec 10 §6).

PermissionDeniedError

PermissionDeniedError(message, *, details=None)

Bases: MineProductivityError

PolicyEngine.evaluate() rejected a Task outright (as opposed to routing it to AWAITING_APPROVAL) -- the one case in this package where a policy violation is not a warning but a hard stop, since dispatching to an Agent lacking the required Permission is never a legitimate outcome to persist as a completed AgentResult (design spec §10, §28).

PolicyConflictError

PolicyConflictError(message, *, details=None)

Bases: RegistrationError

A governance action attempted to re-register an existing, Active AgentPolicy code with a different rule without a version bump -- the Policy-layer analogue of AgentVersionConflictError, mirroring optimization.ProblemConflictError's identical reasoning (spec 10 §6).

TaskNotFoundError

TaskNotFoundError(message, *, details=None)

Bases: NotFoundError

TaskRepository.get(task_id) found no task for that id, or REGISTRY.get(code) found no registered Agent for that code (design spec §6).

TaskExecutor

TaskExecutor(*, repository, policy_engine, audit_trail, retry_policy=None)

Orchestrates one Task: evaluates policy, dispatches to the registered Agent's _act, gates on approval, retries on a recoverable failure, persists the resulting state via the repository swap, and appends terminal outcomes to the audit trail (design spec §12's sequence diagram).

execute

execute(task_id, task, *, context)

Execute task for the instance stored under task_id.

A task already Completed/Failed is terminal (design spec §11): execution is skipped and a warning-carrying result returned. A task stored as AwaitingApproval stays paused -- only :meth:resume with a resolved ApprovalRequest moves it (§16).

Raises:

Type	Description
TaskNotFoundError	If no task is stored under task_id, or no Agent is registered for task.agent_code.
PermissionDeniedError	If PolicyEngine.evaluate denies the task outright -- never persisted as a completed AgentResult (§10, §30).
AgentExecutionError	If the agent's _act failed (retries exhausted or a non-recoverable failure; the task is marked Failed first, §12), or the repository's per-id write serialization contract was violated mid-swap (§32).

resume

resume(task_id, *, approval, context)

Apply an already-resolved approval to the AwaitingApproval task stored under task_id (design spec §16): APPROVED transitions it back to Running and dispatches; REJECTED transitions it directly to Failed, carrying the rejection as a warning on the returned (and audited) AgentResult. This executor never resolves the request itself -- a still-PENDING request is rejected as structurally invalid.

Raises:

Type	Description
TaskNotFoundError	If no task is stored under task_id.
AgentValidationError	If approval is still PENDING, names a different task_id, or the stored task is not AwaitingApproval.

Goal dataclass

Goal(description, *, success_criteria=dict())

Bases: BaseValueObject

A named objective and its success criteria, prior to decomposition into one or more Task\ s (design spec §13).

Examples:

>>> goal = Goal(
...     description="Recover night-shift haulage throughput",
...     success_criteria={"target_tph": 1200.0},
... )
>>> goal.success_criteria["target_tph"]
1200.0

AgentMemory

Bases: ABC

The contract a future memory-backend plugin implements (a vector store, a key-value store, a windowed in-context buffer, or any other recall mechanism). THIS MODULE SHIPS NO CONCRETE SUBCLASS -- choosing a specific embedding model, storage engine, or retention policy is exactly the kind of implementation decision this package's charter (design spec §3.1, §3.5, §4) excludes.

remember abstractmethod

remember(task_id, key, value, *, context)

Persist value under key for task_id.

recall abstractmethod

recall(task_id, key, *, context)

Recall the value stored under key for task_id. A miss returns None and is never an error (design spec §14) -- a calling Agent implementation reasons without whatever was not found.

AgentCategory

Bases: Enum

Closed enum -- adding a member is a governance-reviewed change, mirroring optimization.OptimizationCategory's closed-enum rule (spec 10 §29).

AgentMetadata dataclass

AgentMetadata(code, *, name='', description, tags=frozenset(), attributes=dict(), category, version='1.0.0')

Bases: BaseMetadata

The minimal registration schema for a discoverable Agent type (design spec §29). code names a type (e.g. "FLEET.ReassignmentAdvisor"), never a Task.id.

Examples:

>>> meta = AgentMetadata(
...     code="FLEET.ReassignmentAdvisor",
...     category=AgentCategory.FLEET,
...     description="Advises on truck reassignment after a breakdown.",
... )
>>> meta.name
'FLEET.ReassignmentAdvisor'

AgentPolicy dataclass

AgentPolicy(code, *, version='1.0.0', status=(lambda: PolicyStatus.PROPOSED)(), rule, requires_approval=False, denies=False)

Bases: BaseValueObject

A versioned, governed guardrail on autonomous action (design spec §10). An Active policy is never edited in place: a changed policy is a new version, the prior one Superseded (§10, §26).

Examples:

>>> policy = AgentPolicy(
...     code="POLICY.ShutdownNeedsApproval",
...     rule="capability=approve_shutdown -> require_approval",
...     requires_approval=True,
... )
>>> policy.status
<PolicyStatus.PROPOSED: 'proposed'>

PolicyEngine

Evaluates a Task against the currently Active AgentPolicy set and the assigned Agent's AgentCapabilitySet (design spec §9) before TaskExecutor (§12) dispatches. Returns one of exactly three outcomes, never a fourth (§10, §28).

evaluate

evaluate(task, *, capabilities, policies)

The three-outcome contract (design spec §10): Result.ok(None) (proceed), Result.err carrying the required-approval marker (transition to AWAITING_APPROVAL), or Result.err carrying PermissionDeniedError (a hard stop). Non-Active policies never gate; the caller assembles the candidate policy set (the executor passes every published policy).

PolicyStatus

Bases: Enum

The AgentPolicy lifecycle -- mirrors optimization.ProblemStatus (spec 10 §9) exactly, applied here to governed autonomous-action guardrails.

AgentResult dataclass

AgentResult(task_id='', computed_at=(lambda: datetime.now(timezone.utc))(), warnings=(), *, output=dict(), explanation=None, tool_invocations=())

Bases: BaseValueObject

The shared envelope every concrete agent outcome composes -- mirrors optimization.OptimizationResult's role (spec 10 §18), one layer up. A legitimately incomplete or ambiguous task surfaces as a warning here, never a raise (design spec §30's 'qualify, don't coerce' rule).

Examples:

>>> result = AgentResult(warnings=("no evidence in context",))
>>> result.task_id
''
>>> result.tool_invocations
()

TaskState dataclass

TaskState(attributes)

Bases: BaseValueObject

One task's attributes as of the last executed step -- a frozen value object; the entity is Task (design spec §11). Deliberately not an AgentResult subclass (§20): it represents the task's condition itself, not the outcome of an orchestration call about it.

Examples:

>>> TaskState(attributes={"progress": 0.5}).attributes["progress"]
0.5

Task dataclass

Task(id, goal_code, agent_code, state, status=TaskStatus.SCHEDULED)

Bases: BaseEntity[str]

The root of one executing or completed unit of agent work -- 'Task-as-entity,' following optimization.OptimizationRun's own precedent (spec 10 §10) exactly. Immutable; trivially safe to read and share across threads (design spec §32).

Examples:

>>> task = Task(
...     id="TASK-1",
...     goal_code="GOAL.NightShiftRecovery",
...     agent_code="FLEET.ReassignmentAdvisor",
...     state=TaskState(attributes={"provisioned": True}),
... )
>>> task.status
<TaskStatus.SCHEDULED: 'scheduled'>
>>> task.with_state(task.state, status=TaskStatus.RUNNING).status
<TaskStatus.RUNNING: 'running'>
>>> task.status  # the original instance is untouched
<TaskStatus.SCHEDULED: 'scheduled'>

validate

validate()

Design spec §28: non-empty goal_code and agent_code.

with_state

with_state(state, *, status=None)

Returns a NEW Task instance with state (and optionally status) replacing the current ones -- identical to OptimizationRun.with_state() (spec 10 §10).

TaskStatus

Bases: Enum

A Task's own operational lifecycle -- mirrors optimization.RunStatus (spec 10 §10) in shape, extended with AWAITING_APPROVAL (design spec §11), the one member no lower package's execution lifecycle needed, since none of them pauses for a human decision mid-execution. COMPLETED and FAILED are terminal (§11's state diagram).

Tool

Bases: ABC

The contract a future tool-integration plugin implements (a dispatch-system query, an ERP call, a specific external API). THIS MODULE SHIPS NO CONCRETE SUBCLASS -- choosing a specific external system's integration details is exactly the kind of implementation decision this package's charter (design spec §3.1, §3.5, §4) excludes.

invoke abstractmethod

invoke(*, arguments, context)

Perform this tool's action with arguments, returning its structured result. Implementations validate arguments independently (design spec §33).

ToolInvocation dataclass

ToolInvocation(tool_code, arguments, result, invoked_at)

Bases: BaseValueObject

A record of one Tool.invoke() call and its result, carried on the eventual AgentResult (design spec §20) for audit purposes.

Examples:

>>> from datetime import timezone
>>> invocation = ToolInvocation(
...     tool_code="TOOL.DispatchQuery", arguments={"pit": "north"},
...     result={"trucks": 12}, invoked_at=datetime(2026, 7, 1, tzinfo=timezone.utc),
... )
>>> invocation.result["trucks"]
12

ToolMetadata dataclass

ToolMetadata(code, *, name='', description, tags=frozenset(), attributes=dict(), version='1.0.0')

Bases: BaseMetadata

The minimal registration schema for a discoverable Tool type -- as light as optimization.OptimizationMetadata (design spec §17, §29).

Examples:

>>> meta = ToolMetadata(code="TOOL.DispatchQuery", description="Queries dispatch.")
>>> meta.name
'TOOL.DispatchQuery'

WorkflowEngine

WorkflowEngine(*, executor, repository)

Decomposes a Goal into one or more Task\ s, each assigned to a registered Agent, and coordinates delegation between them (design spec §13, §19) -- the one place a Goal becomes multiple Task\ s.

decompose

decompose(goal, *, context)

Decompose goal into provisioned Task\ s -- one per agent code named by the caller-authored success_criteria["agent_codes"] entry, each persisted into the repository (Scheduled) ready for execution. The first named agent coordinates; each subsequent task's delegation_chain records the handoff (design spec §18).

run

run(goal, *, context)

Decompose goal and execute every resulting Task via the composed TaskExecutor -- sub-tasks dispatch concurrently (design spec §36); results preserve decomposition order regardless of completion order.

register

register(cls)

Register cls into :data:REGISTRY, keyed by cls.meta.code.

Raises:

Type	Description
AgentValidationError	If cls.meta.code is empty (defensive, redundant guard -- AgentMetadata.validate() already rejects it).
AgentVersionConflictError	If cls.meta.code is already registered -- add-only, raised at registration time, never deferred (design spec §26).

register_tool

register_tool(cls)

Register cls into :data:TOOLS, keyed by cls.meta.code -- identical shape and identical error semantics as :func:register, specialized for Tool (design spec §22).

Raises:

Type	Description
AgentValidationError	If cls.meta.code is empty.
AgentVersionConflictError	If cls.meta.code is already registered in :data:TOOLS.

by_category

by_category(category)

A specification satisfied by every task whose registered agent belongs to category -- compose with TaskRepository.list() freely; an empty result is a legitimate answer, never a raise (design spec §23).

Examples:

>>> from mineproductivity.core import InMemoryRepository
>>> repository: InMemoryRepository[Task, str] = InMemoryRepository()
>>> repository.list(by_category(AgentCategory.FLEET))
[]

by_scope

by_scope(scope)

A specification satisfied by every task carrying every key/value pair in scope, resolved against goal_code/ agent_code/status first and the open state.attributes mapping otherwise.

Examples:

>>> from mineproductivity.core import InMemoryRepository
>>> repository: InMemoryRepository[Task, str] = InMemoryRepository()
>>> repository.list(by_scope({"agent_code": "FLEET.ReassignmentAdvisor"}))
[]