Skip to content

API Reference

이 문서는 source code를 직접 열기 전에 "무엇을 구현해야 하는가"를 빠르게 찾기 위한 reference입니다.

Core Models

핵심 도메인 타입:

  • Request: aggregate root
  • Proposal: 사람이 읽는 summary
  • AgentPlan: 실행 가능한 action plan
  • PlannedAction: 단일 실행 action
  • ReviewDecision: plan-level decision
  • ActionReviewDecision: action-level decision
  • PlanReviewResult: review 결과 전체

도메인 규칙은 Domain Model에 설명되어 있습니다. 이 문서는 타입과 확장 지점을 빠르게 찾는 데 집중합니다.

Core Ports

가장 자주 구현하는 포트는 아래 4개입니다.

RequestRepository

  • 역할: Request load/save
  • 구현 예: in-memory, DB, file-based persistence
  • 변경이 필요한 경우: 워크플로 상태를 durable하게 저장해야 할 때

ContextProvider

  • 역할: request와 관련된 외부 context 수집
  • 입력: Request
  • 출력: dict[str, object]
  • 구현 예: DB 조회, API 조회, retrieval, 파일 읽기

AgentEngine

  • 역할: analysis + proposal + optional plan을 한 번에 준비
  • 입력: Request, Mapping[str, object]
  • 출력: AgentWorkResult
  • 구현 예: deterministic engine, LangChain-backed engine

ActionExecutor

  • 역할: review를 통과한 action을 실제로 실행
  • 입력: Request, optional AgentPlan
  • 출력: 실행 결과 문자열
  • 구현 예: fake executor, tool dispatcher, real API client

Workflow Wiring

WorkflowDependencies는 workflow graph가 필요로 하는 외부 경계를 묶습니다.

중요 필드:

  • context_provider
  • agent_engine
  • action_executor
  • requires_human_review
  • auto_reviewer_id

즉 새 앱은 보통 WorkflowDependencies(...)를 구성하는 bootstrap 함수 하나로 시작합니다.

Application DTOs

AnalysisResult

  • structured analysis summary
  • rationale
  • execution steps
  • risk level

AgentWorkResult

  • analysis
  • proposal
  • agent_plan

AgentEngine.prepare(...)는 이 타입을 반환해야 합니다.

ReviewPacket

  • 사람이 review할 때 보게 되는 structured payload
  • request summary, proposal summary, rationale, risk, planned actions 포함

Reference Modules

아래 API는 mkdocstrings로 노출됩니다.

domain_agent.application.ports

Application ports for the generic agent workflow.

ActionExecutor

Bases: Protocol

Boundary for executing the reviewed action.

Source code in src/domain_agent/application/ports.py 
45
46
47
48
49
class ActionExecutor(Protocol):
    """Boundary for executing the reviewed action."""

    def execute(self, request: Request, agent_plan: AgentPlan | None = None) -> str:
        """Execute the request and return execution notes."""

execute(request, agent_plan=None)

Execute the request and return execution notes.

Source code in src/domain_agent/application/ports.py 
48
49
def execute(self, request: Request, agent_plan: AgentPlan | None = None) -> str:
    """Execute the request and return execution notes."""

AgentEngine

Bases: Protocol

Boundary for preparing agent analysis, plan, and proposal output.

Source code in src/domain_agent/application/ports.py 
34
35
36
37
38
39
40
41
42
class AgentEngine(Protocol):
    """Boundary for preparing agent analysis, plan, and proposal output."""

    def prepare(
        self,
        request: Request,
        context: Mapping[str, object],
    ) -> AgentWorkResult:
        """Create structured agent output for the request."""

prepare(request, context)

Create structured agent output for the request.

Source code in src/domain_agent/application/ports.py 
37
38
39
40
41
42
def prepare(
    self,
    request: Request,
    context: Mapping[str, object],
) -> AgentWorkResult:
    """Create structured agent output for the request."""

ContextProvider

Bases: Protocol

External source for request context.

Source code in src/domain_agent/application/ports.py 
27
28
29
30
31
class ContextProvider(Protocol):
    """External source for request context."""

    def get_context(self, request: Request) -> dict[str, object]:
        """Return context relevant to the request."""

get_context(request)

Return context relevant to the request.

Source code in src/domain_agent/application/ports.py 
30
31
def get_context(self, request: Request) -> dict[str, object]:
    """Return context relevant to the request."""

RequestRepository

Bases: Protocol

Persistence boundary for workflow requests.

Source code in src/domain_agent/application/ports.py 
17
18
19
20
21
22
23
24
class RequestRepository(Protocol):
    """Persistence boundary for workflow requests."""

    def get(self, request_id: str) -> Request | None:
        """Load a request by id."""

    def save(self, request: Request) -> None:
        """Persist the current request state."""

get(request_id)

Load a request by id.

Source code in src/domain_agent/application/ports.py 
20
21
def get(self, request_id: str) -> Request | None:
    """Load a request by id."""

save(request)

Persist the current request state.

Source code in src/domain_agent/application/ports.py 
23
24
def save(self, request: Request) -> None:
    """Persist the current request state."""

Tool

Bases: Protocol

Executable tool used by a planned action.

Source code in src/domain_agent/application/ports.py 
52
53
54
55
56
57
58
class Tool(Protocol):
    """Executable tool used by a planned action."""

    name: str

    def execute(self, request: Request, action: PlannedAction) -> str:
        """Execute a planned action for the request."""

execute(request, action)

Execute a planned action for the request.

Source code in src/domain_agent/application/ports.py 
57
58
def execute(self, request: Request, action: PlannedAction) -> str:
    """Execute a planned action for the request."""

ToolRegistry

Bases: Protocol

Lookup boundary for executable tools.

Source code in src/domain_agent/application/ports.py 
61
62
63
64
65
class ToolRegistry(Protocol):
    """Lookup boundary for executable tools."""

    def get(self, tool_name: str) -> Tool:
        """Return the registered tool for the given name."""

get(tool_name)

Return the registered tool for the given name.

Source code in src/domain_agent/application/ports.py 
64
65
def get(self, tool_name: str) -> Tool:
    """Return the registered tool for the given name."""

WorkflowDependencies dataclass

Dependencies required by the workflow graph.

Source code in src/domain_agent/application/ports.py 
68
69
70
71
72
73
74
75
76
@dataclass(frozen=True)
class WorkflowDependencies:
    """Dependencies required by the workflow graph."""

    context_provider: ContextProvider
    agent_engine: AgentEngine
    action_executor: ActionExecutor
    requires_human_review: bool = True
    auto_reviewer_id: str = "workflow-system"

domain_agent.application.dto

Application-layer data structures.

AgentWorkResult dataclass

Structured output returned by an agent engine.

Source code in src/domain_agent/application/dto.py 
36
37
38
39
40
41
42
43
@dataclass(frozen=True)
class AgentWorkResult:
    """Structured output returned by an agent engine."""

    analysis: AnalysisResult
    proposal: Proposal
    agent_plan: AgentPlan | None = None
    observations: tuple[Observation, ...] = ()

AnalysisResult dataclass

Structured analysis output used to build a proposal.

Source code in src/domain_agent/application/dto.py 
17
18
19
20
21
22
23
24
@dataclass(frozen=True)
class AnalysisResult:
    """Structured analysis output used to build a proposal."""

    summary: str
    rationale: str
    execution_steps: tuple[str, ...]
    risk_level: RiskLevel

Observation dataclass

A compact record of evidence gathered during agent preparation.

Source code in src/domain_agent/application/dto.py 
27
28
29
30
31
32
33
@dataclass(frozen=True)
class Observation:
    """A compact record of evidence gathered during agent preparation."""

    source: str
    summary: str
    details: str = ""

ReviewPacket dataclass

Structured payload presented to a human reviewer.

Source code in src/domain_agent/application/dto.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
@dataclass(frozen=True)
class ReviewPacket:
    """Structured payload presented to a human reviewer."""

    request_id: str
    requester_id: str
    request_summary: str
    proposal_summary: str
    rationale: str
    risk_level: RiskLevel
    execution_steps: tuple[str, ...]
    planned_actions: tuple[PlannedAction, ...]
    observations: tuple[Observation, ...] = ()
    action_decisions: tuple[ActionReviewDecision, ...] = ()

build_review_packet(request, proposal, *, observations=())

Create a structured review packet from the request and proposal.

Source code in src/domain_agent/application/dto.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def build_review_packet(
    request: Request,
    proposal: Proposal,
    *,
    observations: tuple[Observation, ...] = (),
) -> ReviewPacket:
    """Create a structured review packet from the request and proposal."""
    planned_actions: tuple[PlannedAction, ...]
    if proposal.agent_plan is not None:
        planned_actions = proposal.agent_plan.actions
    else:
        planned_actions = (
            PlannedAction(
                action_id="plan-review",
                tool_name="plan_review",
                description="Review the overall proposal as a single action.",
            ),
        )

    return ReviewPacket(
        request_id=request.request_id,
        requester_id=request.requester_id,
        request_summary=request.summary,
        proposal_summary=proposal.summary,
        rationale=proposal.rationale,
        risk_level=proposal.risk_level,
        execution_steps=proposal.execution_steps,
        planned_actions=planned_actions,
        observations=observations,
    )

domain_agent.application.use_cases

Application use cases for orchestrating agent workflows.

AgentWorkflowService dataclass

Orchestrate workflow execution through application ports.

Source code in src/domain_agent/application/use_cases.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
@dataclass(frozen=True)
class AgentWorkflowService:
    """Orchestrate workflow execution through application ports."""

    request_repository: RequestRepository
    workflow_dependencies: WorkflowDependencies
    workflow_shape: WorkflowShape = DEFAULT_WORKFLOW_SHAPE

    def run(
        self,
        request_id: str,
        *,
        review_handler: ReviewHandler | None = None,
    ) -> WorkflowState:
        """Load a request, execute the workflow, and persist the result."""
        request = self.request_repository.get(request_id)
        if request is None:
            raise RequestNotFoundError(f"Request '{request_id}' was not found.")

        workflow = build_workflow(
            self.workflow_dependencies,
            workflow_shape=self.workflow_shape,
            checkpointer=build_in_memory_checkpointer(),
        )
        config = {"configurable": {"thread_id": request.request_id}}
        next_input: object = initial_workflow_state(request)

        while True:
            result = cast(dict[str, Any], workflow.invoke(next_input, config=config))
            interrupts = cast(tuple[Interrupt, ...], tuple(result.get("__interrupt__", ())))
            if not interrupts:
                final_result = cast(WorkflowState, result)
                self.request_repository.save(final_result["request"])
                return final_result

            if review_handler is None:
                raise ReviewHandlerRequiredError(
                    "Workflow paused for review, but no review_handler was provided."
                )

            interrupt_value = interrupts[0].value
            if not isinstance(interrupt_value, ReviewPacket):
                raise TypeError("Workflow interrupt value must be a ReviewPacket.")

            next_input = Command(
                resume=review_handler(interrupt_value),
                update={"request": result.get("request")},
            )

run(request_id, *, review_handler=None)

Load a request, execute the workflow, and persist the result.

Source code in src/domain_agent/application/use_cases.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def run(
    self,
    request_id: str,
    *,
    review_handler: ReviewHandler | None = None,
) -> WorkflowState:
    """Load a request, execute the workflow, and persist the result."""
    request = self.request_repository.get(request_id)
    if request is None:
        raise RequestNotFoundError(f"Request '{request_id}' was not found.")

    workflow = build_workflow(
        self.workflow_dependencies,
        workflow_shape=self.workflow_shape,
        checkpointer=build_in_memory_checkpointer(),
    )
    config = {"configurable": {"thread_id": request.request_id}}
    next_input: object = initial_workflow_state(request)

    while True:
        result = cast(dict[str, Any], workflow.invoke(next_input, config=config))
        interrupts = cast(tuple[Interrupt, ...], tuple(result.get("__interrupt__", ())))
        if not interrupts:
            final_result = cast(WorkflowState, result)
            self.request_repository.save(final_result["request"])
            return final_result

        if review_handler is None:
            raise ReviewHandlerRequiredError(
                "Workflow paused for review, but no review_handler was provided."
            )

        interrupt_value = interrupts[0].value
        if not isinstance(interrupt_value, ReviewPacket):
            raise TypeError("Workflow interrupt value must be a ReviewPacket.")

        next_input = Command(
            resume=review_handler(interrupt_value),
            update={"request": result.get("request")},
        )

RequestNotFoundError

Bases: LookupError

Raised when a request cannot be found in the repository.

Source code in src/domain_agent/application/use_cases.py 
23
24
class RequestNotFoundError(LookupError):
    """Raised when a request cannot be found in the repository."""

ReviewHandlerRequiredError

Bases: RuntimeError

Raised when workflow execution pauses for review without a handler.

Source code in src/domain_agent/application/use_cases.py 
27
28
class ReviewHandlerRequiredError(RuntimeError):
    """Raised when workflow execution pauses for review without a handler."""

domain_agent.domain.models

Core domain model for a generic agent workflow.

ActionReviewDecision dataclass

A human decision recorded for a single planned action.

Source code in src/domain_agent/domain/models.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
@dataclass(frozen=True)
class ActionReviewDecision:
    """A human decision recorded for a single planned action."""

    action_id: str
    approved: bool
    comment: str = ""

    def __post_init__(self) -> None:
        if not self.action_id.strip():
            raise DomainError("Action review action_id must not be empty.")
        if not self.approved and not self.comment.strip():
            raise DomainError("Rejected action reviews must include a comment.")

AgentPlan dataclass

A structured plan of executable actions prepared by the agent.

Source code in src/domain_agent/domain/models.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
@dataclass(frozen=True)
class AgentPlan:
    """A structured plan of executable actions prepared by the agent."""

    summary: str
    rationale: str
    actions: tuple[PlannedAction, ...]
    expected_outcome: str
    risk_level: RiskLevel

    def __post_init__(self) -> None:
        if not self.summary.strip():
            raise DomainError("Agent plan summary must not be empty.")
        if not self.rationale.strip():
            raise DomainError("Agent plan rationale must not be empty.")
        if not self.expected_outcome.strip():
            raise DomainError("Agent plan expected_outcome must not be empty.")
        if not self.actions:
            raise DomainError("Agent plan must include at least one planned action.")

DomainError

Bases: ValueError

Base error for invalid domain operations.

Source code in src/domain_agent/domain/models.py 
10
11
class DomainError(ValueError):
    """Base error for invalid domain operations."""

InvalidStatusTransitionError

Bases: DomainError

Raised when a request attempts an invalid lifecycle transition.

Source code in src/domain_agent/domain/models.py 
14
15
class InvalidStatusTransitionError(DomainError):
    """Raised when a request attempts an invalid lifecycle transition."""

PlanReviewResult dataclass

Structured review outcome for a full plan review.

Source code in src/domain_agent/domain/models.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
@dataclass(frozen=True)
class PlanReviewResult:
    """Structured review outcome for a full plan review."""

    decision: ReviewDecision
    action_decisions: tuple[ActionReviewDecision, ...]

    def __post_init__(self) -> None:
        if not self.action_decisions:
            raise DomainError("Plan review result must include action decisions.")
        approved_actions = tuple(
            action_decision for action_decision in self.action_decisions if action_decision.approved
        )
        if self.decision.approved and not approved_actions:
            raise DomainError("A positive plan review requires at least one approved action.")
        if not self.decision.approved and approved_actions:
            raise DomainError("Rejected plans cannot contain approved actions.")

PlannedAction dataclass

A single tool-backed action planned by the agent.

Source code in src/domain_agent/domain/models.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
@dataclass(frozen=True)
class PlannedAction:
    """A single tool-backed action planned by the agent."""

    action_id: str
    tool_name: str
    description: str
    arguments: dict[str, object] = field(default_factory=dict)

    def __post_init__(self) -> None:
        if not self.action_id.strip():
            raise DomainError("Planned action action_id must not be empty.")
        if not self.tool_name.strip():
            raise DomainError("Planned action tool_name must not be empty.")
        if not self.description.strip():
            raise DomainError("Planned action description must not be empty.")

        object.__setattr__(self, "arguments", dict(self.arguments))

Proposal dataclass

A human-reviewable plan prepared before execution.

Source code in src/domain_agent/domain/models.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
@dataclass(frozen=True)
class Proposal:
    """A human-reviewable plan prepared before execution."""

    summary: str
    rationale: str
    execution_steps: tuple[str, ...]
    risk_level: RiskLevel
    prepared_by: str
    agent_plan: AgentPlan | None = None

    def __post_init__(self) -> None:
        if not self.summary.strip():
            raise DomainError("Proposal summary must not be empty.")
        if not self.rationale.strip():
            raise DomainError("Proposal rationale must not be empty.")
        if not self.prepared_by.strip():
            raise DomainError("Proposal prepared_by must not be empty.")
        if not self.execution_steps:
            raise DomainError("Proposal must include at least one execution step.")
        if any(not step.strip() for step in self.execution_steps):
            raise DomainError("Proposal execution steps must not be empty.")

Request dataclass

Aggregate root for a reviewable request.

Source code in src/domain_agent/domain/models.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
@dataclass
class Request:
    """Aggregate root for a reviewable request."""

    request_id: str
    requester_id: str
    summary: str
    risk_level: RiskLevel
    details: str = ""
    status: RequestStatus = RequestStatus.DRAFT
    proposal: Proposal | None = None
    review_decision: ReviewDecision | None = None
    action_review_decisions: tuple[ActionReviewDecision, ...] = ()
    execution_notes: str | None = None
    _executed: bool = field(init=False, default=False, repr=False)

    def __post_init__(self) -> None:
        if not self.request_id.strip():
            raise DomainError("Request request_id must not be empty.")
        if not self.requester_id.strip():
            raise DomainError("Request requester_id must not be empty.")
        if not self.summary.strip():
            raise DomainError("Request summary must not be empty.")

        if self.status in {
            RequestStatus.PROPOSED,
            RequestStatus.AWAITING_APPROVAL,
            RequestStatus.APPROVED,
            RequestStatus.REJECTED,
            RequestStatus.EXECUTED,
        } and self.proposal is None:
            raise DomainError(f"Request in status {self.status.value} requires a proposal.")

        if self.status in {
            RequestStatus.APPROVED,
            RequestStatus.REJECTED,
            RequestStatus.EXECUTED,
        } and self.review_decision is None:
            raise DomainError(
                f"Request in status {self.status.value} requires a review decision."
            )

        if self.status is RequestStatus.REJECTED and self.review_decision is not None:
            if self.review_decision.approved:
                raise DomainError("Rejected request cannot carry an approving decision.")

        if (
            self.status in {RequestStatus.APPROVED, RequestStatus.EXECUTED}
            and self.review_decision is not None
        ):
            if not self.review_decision.approved:
                raise DomainError("Approved or executed request requires an approving decision.")

        if self.action_review_decisions:
            if self.proposal is None:
                raise DomainError("Action reviews require a proposal.")
            planned_action_ids = self._expected_action_ids()
            decision_action_ids = {
                action_decision.action_id for action_decision in self.action_review_decisions
            }
            if planned_action_ids != decision_action_ids:
                raise DomainError("Action reviews must cover each planned action exactly once.")
        elif (
            self.status in {RequestStatus.APPROVED, RequestStatus.REJECTED, RequestStatus.EXECUTED}
            and self.proposal is not None
        ):
            raise DomainError("Approved, rejected, or executed requests require action reviews.")

        self._executed = self.status is RequestStatus.EXECUTED

    def start_review(self) -> None:
        """Move a draft request into review."""
        self._transition_to(RequestStatus.UNDER_REVIEW)

    def attach_proposal(self, proposal: Proposal) -> None:
        """Attach a proposal after analysis has completed."""
        if self.status is not RequestStatus.UNDER_REVIEW:
            raise InvalidStatusTransitionError(
                "Proposal can only be attached while the request is under review."
            )
        self.proposal = proposal
        self._transition_to(RequestStatus.PROPOSED)

    def request_review(self) -> None:
        """Expose the prepared proposal for human review."""
        if self.status is not RequestStatus.PROPOSED:
            raise InvalidStatusTransitionError(
                "Review can only be requested after a proposal has been prepared."
            )
        if self.proposal is None:
            raise DomainError("Review cannot be requested without a proposal.")
        self._transition_to(RequestStatus.AWAITING_APPROVAL)

    def record_review_result(self, review_result: PlanReviewResult) -> None:
        """Record the human plan review result and update the request state."""
        if self.status is not RequestStatus.AWAITING_APPROVAL:
            raise InvalidStatusTransitionError(
                "Review decisions can only be recorded while awaiting review."
            )
        if self.proposal is None:
            raise DomainError("Action-level review requires a proposal.")

        planned_action_ids = self._expected_action_ids()
        decision_action_ids = {
            action_decision.action_id for action_decision in review_result.action_decisions
        }
        if planned_action_ids != decision_action_ids:
            raise DomainError("Review result must include a decision for every planned action.")

        self.review_decision = review_result.decision
        self.action_review_decisions = review_result.action_decisions
        target_status = (
            RequestStatus.APPROVED if review_result.decision.approved else RequestStatus.REJECTED
        )
        self._transition_to(target_status)

    def approved_action_ids(self) -> tuple[str, ...]:
        """Return the action ids positively reviewed for execution."""
        return tuple(
            action_decision.action_id
            for action_decision in self.action_review_decisions
            if action_decision.approved
        )

    def _expected_action_ids(self) -> set[str]:
        if self.proposal is None:
            raise DomainError("Expected action ids require a proposal.")
        if self.proposal.agent_plan is None:
            return {"plan-review"}
        return {planned_action.action_id for planned_action in self.proposal.agent_plan.actions}

    def execute(self, execution_notes: str | None = None) -> None:
        """Mark the reviewed request as executed."""
        if self.status is RequestStatus.REJECTED:
            raise ReviewRequiredError("Rejected requests cannot be executed.")
        if self.status is not RequestStatus.APPROVED:
            raise ReviewRequiredError("Review is required before execution.")
        if self.review_decision is None or not self.review_decision.approved:
            raise ReviewRequiredError("A positive review decision is required before execution.")

        self.execution_notes = execution_notes
        self._transition_to(RequestStatus.EXECUTED)
        self._executed = True

    def _transition_to(self, target_status: RequestStatus) -> None:
        if not self.status.can_transition_to(target_status):
            raise InvalidStatusTransitionError(
                f"Cannot transition request from {self.status.value} to {target_status.value}."
            )
        self.status = target_status

approved_action_ids()

Return the action ids positively reviewed for execution.

Source code in src/domain_agent/domain/models.py 
292
293
294
295
296
297
298
def approved_action_ids(self) -> tuple[str, ...]:
    """Return the action ids positively reviewed for execution."""
    return tuple(
        action_decision.action_id
        for action_decision in self.action_review_decisions
        if action_decision.approved
    )

attach_proposal(proposal)

Attach a proposal after analysis has completed.

Source code in src/domain_agent/domain/models.py 
250
251
252
253
254
255
256
257
def attach_proposal(self, proposal: Proposal) -> None:
    """Attach a proposal after analysis has completed."""
    if self.status is not RequestStatus.UNDER_REVIEW:
        raise InvalidStatusTransitionError(
            "Proposal can only be attached while the request is under review."
        )
    self.proposal = proposal
    self._transition_to(RequestStatus.PROPOSED)

execute(execution_notes=None)

Mark the reviewed request as executed.

Source code in src/domain_agent/domain/models.py 
307
308
309
310
311
312
313
314
315
316
317
318
def execute(self, execution_notes: str | None = None) -> None:
    """Mark the reviewed request as executed."""
    if self.status is RequestStatus.REJECTED:
        raise ReviewRequiredError("Rejected requests cannot be executed.")
    if self.status is not RequestStatus.APPROVED:
        raise ReviewRequiredError("Review is required before execution.")
    if self.review_decision is None or not self.review_decision.approved:
        raise ReviewRequiredError("A positive review decision is required before execution.")

    self.execution_notes = execution_notes
    self._transition_to(RequestStatus.EXECUTED)
    self._executed = True

record_review_result(review_result)

Record the human plan review result and update the request state.

Source code in src/domain_agent/domain/models.py 
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
def record_review_result(self, review_result: PlanReviewResult) -> None:
    """Record the human plan review result and update the request state."""
    if self.status is not RequestStatus.AWAITING_APPROVAL:
        raise InvalidStatusTransitionError(
            "Review decisions can only be recorded while awaiting review."
        )
    if self.proposal is None:
        raise DomainError("Action-level review requires a proposal.")

    planned_action_ids = self._expected_action_ids()
    decision_action_ids = {
        action_decision.action_id for action_decision in review_result.action_decisions
    }
    if planned_action_ids != decision_action_ids:
        raise DomainError("Review result must include a decision for every planned action.")

    self.review_decision = review_result.decision
    self.action_review_decisions = review_result.action_decisions
    target_status = (
        RequestStatus.APPROVED if review_result.decision.approved else RequestStatus.REJECTED
    )
    self._transition_to(target_status)

request_review()

Expose the prepared proposal for human review.

Source code in src/domain_agent/domain/models.py 
259
260
261
262
263
264
265
266
267
def request_review(self) -> None:
    """Expose the prepared proposal for human review."""
    if self.status is not RequestStatus.PROPOSED:
        raise InvalidStatusTransitionError(
            "Review can only be requested after a proposal has been prepared."
        )
    if self.proposal is None:
        raise DomainError("Review cannot be requested without a proposal.")
    self._transition_to(RequestStatus.AWAITING_APPROVAL)

start_review()

Move a draft request into review.

Source code in src/domain_agent/domain/models.py 
246
247
248
def start_review(self) -> None:
    """Move a draft request into review."""
    self._transition_to(RequestStatus.UNDER_REVIEW)

RequestStatus

Bases: StrEnum

Lifecycle states for a reviewable request.

Source code in src/domain_agent/domain/models.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
class RequestStatus(StrEnum):
    """Lifecycle states for a reviewable request."""

    DRAFT = "draft"
    UNDER_REVIEW = "under_review"
    PROPOSED = "proposed"
    AWAITING_APPROVAL = "awaiting_approval"
    APPROVED = "approved"
    REJECTED = "rejected"
    EXECUTED = "executed"

    def can_transition_to(self, target: RequestStatus) -> bool:
        """Return whether this status can move directly to the target status."""
        allowed_transitions: dict[RequestStatus, set[RequestStatus]] = {
            RequestStatus.DRAFT: {RequestStatus.UNDER_REVIEW},
            RequestStatus.UNDER_REVIEW: {RequestStatus.PROPOSED},
            RequestStatus.PROPOSED: {RequestStatus.AWAITING_APPROVAL},
            RequestStatus.AWAITING_APPROVAL: {
                RequestStatus.APPROVED,
                RequestStatus.REJECTED,
            },
            RequestStatus.APPROVED: {RequestStatus.EXECUTED},
            RequestStatus.REJECTED: set(),
            RequestStatus.EXECUTED: set(),
        }
        return target in allowed_transitions[self]

can_transition_to(target)

Return whether this status can move directly to the target status.

Source code in src/domain_agent/domain/models.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def can_transition_to(self, target: RequestStatus) -> bool:
    """Return whether this status can move directly to the target status."""
    allowed_transitions: dict[RequestStatus, set[RequestStatus]] = {
        RequestStatus.DRAFT: {RequestStatus.UNDER_REVIEW},
        RequestStatus.UNDER_REVIEW: {RequestStatus.PROPOSED},
        RequestStatus.PROPOSED: {RequestStatus.AWAITING_APPROVAL},
        RequestStatus.AWAITING_APPROVAL: {
            RequestStatus.APPROVED,
            RequestStatus.REJECTED,
        },
        RequestStatus.APPROVED: {RequestStatus.EXECUTED},
        RequestStatus.REJECTED: set(),
        RequestStatus.EXECUTED: set(),
    }
    return target in allowed_transitions[self]

ReviewDecision dataclass

A human decision recorded against a proposed request.

Source code in src/domain_agent/domain/models.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
@dataclass(frozen=True)
class ReviewDecision:
    """A human decision recorded against a proposed request."""

    approved: bool
    decided_by: str
    decided_at: dt.datetime
    comment: str = ""

    def __post_init__(self) -> None:
        if not self.decided_by.strip():
            raise DomainError("Review decision decided_by must not be empty.")
        if self.decided_at.tzinfo is None:
            raise DomainError("Review decision decided_at must be timezone-aware.")
        if not self.approved and not self.comment.strip():
            raise DomainError("Rejected decisions must include a comment.")

ReviewRequiredError

Bases: DomainError

Raised when execution is attempted before a required review outcome.

Source code in src/domain_agent/domain/models.py 
18
19
class ReviewRequiredError(DomainError):
    """Raised when execution is attempted before a required review outcome."""

RiskLevel

Bases: StrEnum

Generic risk classification for a request or proposal.

Source code in src/domain_agent/domain/models.py 
50
51
52
53
54
55
56
class RiskLevel(StrEnum):
    """Generic risk classification for a request or proposal."""

    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"