Skip to content

[FEAT] Add stateful responses layer with history rehydration and DB persistence #21

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
maralbahari wants to merge 16 commits into from
Closed

[FEAT] Add stateful responses layer with history rehydration and DB persistence #21

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
16 commits
Select commit Hold shift + click to select a range
9f043b5
add databse and stateful responses flow
maralbahari Apr 14, 2026
c46549c
update rehydration with conversation table
maralbahari Apr 14, 2026
42fd05b
add unit tests
maralbahari Apr 14, 2026
73da713
make conversation api presist along with responses
maralbahari Apr 16, 2026
4b06d40
clean code
maralbahari Apr 17, 2026
7eddfa5
update tests
maralbahari Apr 17, 2026
1b182f7
update packages depencencies
maralbahari Apr 17, 2026
adafa76
fix unit tests and add cassettes with recoder
maralbahari Apr 21, 2026
cf067bf
Merge remote-tracking branch 'origin/main' into responses-flow
maralbahari Apr 21, 2026
41aa568
get dialect from db url and add validators
maralbahari Apr 27, 2026
dd462b2
fix db url validator
maralbahari May 4, 2026
1120a5f
redesign conversation db and resolve Critical: Lost-update race in pu…
maralbahari May 4, 2026
373ea72
session_factory_engine unconditional replacement
maralbahari May 4, 2026
eeafb8b
fix exception catch and missing event in sse
maralbahari May 4, 2026
d0c6b74
add todo comment
maralbahari May 4, 2026
155e8b1
add TODO for proxy router
maralbahari May 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions .gitignore
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,3 +16,10 @@ build/
.vscode/
*.swp
*.swo

# DB
*.db
*.db-shm
*db-wal

*.log
2 changes: 1 addition & 1 deletion .pre-commit-config.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
exclude: 'build/'
exclude: 'build/|^tests/cassettes/text_only/'
minimum_pre_commit_version: 4.4.0
x-uv-dependency: &uv-dependency "uv==0.9.15"
default_language_version:
Expand Down
10 changes: 10 additions & 0 deletions pyproject.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,16 @@ dependencies = [
"uvicorn[standard]>=0.30.0",
"httpx>=0.27.0",
"pydantic>=2.7.0",
"pydantic-ai-slim>=0.1.0",
"sqlalchemy[asyncio]>=2.0.0",
"aiosqlite>=0.19.0",
"uuid-utils>=0.9.0",
"openai>=2.32.0",
]

[project.optional-dependencies]
postgresql = [
"psycopg[binary]>=3.1.0",
]

[project.scripts]
Expand Down
40 changes: 39 additions & 1 deletion src/agentic_api/config/runtime.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,7 @@
from pydantic import BaseModel, Field
from typing import Literal

from pydantic import BaseModel, Field, computed_field, field_validator
from sqlalchemy.engine import make_url


class RuntimeConfig(BaseModel):
Expand All @@ -19,3 +22,38 @@ class RuntimeConfig(BaseModel):
upstream_ready_interval_s: float = Field(
description="Polling interval in seconds when waiting for the upstream to become ready."
)

db_url: str = Field(
default="sqlite+aiosqlite:///./agentic_api.db",
description="SQLAlchemy async database URL.",
)

@field_validator("db_url")
@classmethod
def validate_db_url(cls, v: str) -> str:
url = make_url(v)
dialect = url.get_dialect()
if not dialect.is_async:
raise ValueError(
f"db_url must use an async driver (e.g. sqlite+aiosqlite://, postgresql+asyncpg://), got: {v!r}"
)
if dialect.name not in ("sqlite", "postgresql"):
raise ValueError(
f"db_url dialect must be sqlite or postgresql, got: {dialect.name!r}"
)
return v

@computed_field
@property
def db_dialect(self) -> Literal["sqlite", "postgresql"]:
return make_url(self.db_url).get_dialect().name

@franciscojavierarceo franciscojavierarceo Apr 29, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

High: db_dialect type annotation is misleading

db_dialect is typed as Literal["sqlite", "postgresql"] but the validate_db_url validator only checks is_async, not the dialect name. A URL like mysql+aiomysql://... passes validation but violates the return type.

Suggestion — validate the dialect name explicitly in validate_db_url:

dialect_name = url.get_dialect().name
if dialect_name not in ("sqlite", "postgresql"):
    raise ValueError(f"Unsupported dialect: {dialect_name}. Only sqlite and postgresql are supported.")

This also addresses the open review thread from @noobHappylife — db_dialect can then be fully derived from db_url with confidence.

@maralbahari maralbahari May 4, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@franciscojavierarceo I have included the validation in db_url pydantic field validator so that user only enters sqlite or postgresql in the first place. since db_dialect is a computed field based on db_url then we are sure that we only return intended dialect.


response_store_enabled: bool = Field(
default=True,
description="Enable response persistence for multi-turn rehydration.",
)

log_model_messages: bool = Field(
default=False,
description="Log full pydantic_ai model messages on engine failure (verbose).",
)
309 changes: 309 additions & 0 deletions src/agentic_api/core/composer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,309 @@
from __future__ import annotations

from collections.abc import Iterable
from dataclasses import dataclass
from time import time

from agentic_api.core.normalized_events import (
FunctionCallArgumentsDelta,
FunctionCallDone,
FunctionCallStarted,
MessageDelta,
MessageDone,
MessageStarted,
NormalizedEvent,
ReasoningDelta,
ReasoningDone,
ReasoningStarted,
UsageFinal,
)
from agentic_api.types.responses import (
ContentPartEvent,
ErrorEvent,
FunctionToolCall,
IncompleteDetails,
InputTokenDetails,
OutputItemEvent,
OutputMessage,
OutputTextContent,
OutputTokenDetails,
ResponseEvent,
ResponseUsage,
ResponsesResponse,
StreamEvent,
TextDeltaEvent,
)
from agentic_api.utils.common import uuid7_str


@dataclass(slots=True)
class _ItemState:
item_id: str
output_index: int
kind: str
text: str = ""
reasoning: str = ""
function_name: str | None = None
function_call_id: str | None = None
function_args_json: str = ""


class ResponseComposer:
"""Compose a Responses SSE/JSON contract from NormalizedEvents.

Handles: text messages, reasoning, and user-defined function tool calls.
"""

def __init__(
self, *, response: ResponsesResponse, include: set[str] | None = None
) -> None:
self._response = response
self._started = False
self._sequence_number = 0
self._next_output_index = 0
self._items: dict[str, _ItemState] = {}
self._output_items: list[OutputMessage | FunctionToolCall] = []

@property
def response(self) -> ResponsesResponse:
return self._response

def feed(self, event: NormalizedEvent) -> Iterable[StreamEvent]:
if not self._started:
raise RuntimeError("ResponseComposer.start() must be called before feed().")

if isinstance(event, MessageStarted):
yield from self._start_message(event)
elif isinstance(event, MessageDelta):
yield from self._message_delta(event)
elif isinstance(event, MessageDone):
yield from self._message_done(event)
elif isinstance(event, ReasoningStarted):
pass # Reasoning items are not emitted as output events in this implementation

@noobHappylife noobHappylife Apr 23, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

what does this means? not emitted as output events

@maralbahari maralbahari Apr 23, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes for now. This PR doesnt focus on that. in another PR the reasoning events would be added.

elif isinstance(event, ReasoningDelta):
pass
elif isinstance(event, ReasoningDone):
pass
elif isinstance(event, FunctionCallStarted):
yield from self._start_function_call(event)
elif isinstance(event, FunctionCallArgumentsDelta):
yield from self._function_args_delta(event)
elif isinstance(event, FunctionCallDone):
yield from self._function_done(event)
elif isinstance(event, UsageFinal):
yield from self._complete_response(event)

def start(self) -> Iterable[StreamEvent]:
if self._started:
return []
self._started = True
self._response.status = "in_progress"
return [
ResponseEvent(
type="response.created",
sequence_number=self.alloc_sequence_number(),
response=self._response,
),
ResponseEvent(
type="response.in_progress",
sequence_number=self.alloc_sequence_number(),
response=self._response,
),
]

def alloc_sequence_number(self) -> int:
return self._incr_seq()

def _incr_seq(self) -> int:
current = self._sequence_number
self._sequence_number += 1
return current

def _alloc_output_index(self) -> int:
current = self._next_output_index
self._next_output_index += 1
return current

def _start_message(self, event: MessageStarted) -> Iterable[StreamEvent]:
item_id = uuid7_str("msg_")
out_index = self._alloc_output_index()
state = _ItemState(item_id=item_id, output_index=out_index, kind="message")
self._items[event.item_key] = state

yield OutputItemEvent(
type="response.output_item.added",
sequence_number=self._incr_seq(),
output_index=out_index,
item=OutputMessage(content=[], id=item_id, status="in_progress"),
)
yield ContentPartEvent(
type="response.content_part.added",
item_id=item_id,
sequence_number=self._incr_seq(),
output_index=out_index,
content_index=0,
part=OutputTextContent(text=""),
)

def _message_delta(self, event: MessageDelta) -> Iterable[StreamEvent]:
state = self._items[event.item_key]
state.text += event.delta
yield TextDeltaEvent(
type="response.output_text.delta",
item_id=state.item_id,
sequence_number=self._incr_seq(),
output_index=state.output_index,
content_index=0,
delta=event.delta,
logprobs=[],
)

def _message_done(self, event: MessageDone) -> Iterable[StreamEvent]:
state = self._items[event.item_key]
state.text = event.text
yield TextDeltaEvent(
type="response.output_text.done",
item_id=state.item_id,
sequence_number=self._incr_seq(),
output_index=state.output_index,
content_index=0,
text=event.text,
logprobs=[],
)
yield ContentPartEvent(
type="response.content_part.done",
item_id=state.item_id,
sequence_number=self._incr_seq(),
output_index=state.output_index,
content_index=0,
part=OutputTextContent(text=event.text),
)
item = OutputMessage(
content=[OutputTextContent(text=event.text)],
id=state.item_id,
status="completed",
)
yield OutputItemEvent(
type="response.output_item.done",
sequence_number=self._incr_seq(),
output_index=state.output_index,
item=item,
)
self._output_items.append(item)

def _start_function_call(self, event: FunctionCallStarted) -> Iterable[StreamEvent]:
item_id = uuid7_str("fc_")
out_index = self._alloc_output_index()
state = _ItemState(
item_id=item_id,
output_index=out_index,
kind="function_call",
function_name=event.name,
function_call_id=event.call_id,
function_args_json=event.initial_arguments_json,
)
self._items[event.item_key] = state

yield OutputItemEvent(
type="response.output_item.added",
sequence_number=self._incr_seq(),
output_index=out_index,
item=FunctionToolCall(
arguments=event.initial_arguments_json,
call_id=event.call_id,
name=event.name,
id=item_id,
status="in_progress",
),
)

def _function_args_delta(
self, event: FunctionCallArgumentsDelta
) -> Iterable[StreamEvent]:
state = self._items[event.item_key]
state.function_args_json += event.delta
yield TextDeltaEvent(
type="response.function_call_arguments.delta",
item_id=state.item_id,
sequence_number=self._incr_seq(),
output_index=state.output_index,
delta=event.delta,
)

def _function_done(self, event: FunctionCallDone) -> Iterable[StreamEvent]:
state = self._items[event.item_key]
state.function_args_json = event.arguments_json
yield TextDeltaEvent(
type="response.function_call_arguments.done",
item_id=state.item_id,
sequence_number=self._incr_seq(),
output_index=state.output_index,
arguments=event.arguments_json,
)
item = FunctionToolCall(
arguments=event.arguments_json,
call_id=state.function_call_id or "",
name=state.function_name or "",
id=state.item_id,
status="completed",
)
yield OutputItemEvent(
type="response.output_item.done",
sequence_number=self._incr_seq(),
output_index=state.output_index,
item=item,
)
self._output_items.append(item)

def _complete_response(self, event: UsageFinal) -> Iterable[StreamEvent]:
self._response.usage = ResponseUsage(
input_tokens=event.input_tokens,
input_tokens_details=InputTokenDetails(
cached_tokens=event.cache_read_tokens
),
output_tokens=event.output_tokens,
output_tokens_details=OutputTokenDetails(
reasoning_tokens=event.reasoning_tokens
),
total_tokens=event.total_tokens,
)
self._response.output = list(self._output_items)
if event.incomplete_reason is not None:
self._response.status = "incomplete"
self._response.incomplete_details = IncompleteDetails(
reason=event.incomplete_reason
)
yield ResponseEvent(
type="response.incomplete",
sequence_number=self._incr_seq(),
response=self._response,
)
return

self._response.status = "completed"
self._response.incomplete_details = None
self._response.created_at = int(time())
yield ResponseEvent(
type="response.completed",
sequence_number=self._incr_seq(),
response=self._response,
)

def make_error_events(
self, *, code: str, message: str, param: str | None = None
) -> Iterable[StreamEvent]:
"""Emit an error event + response.failed lifecycle event."""
self._response.error = {"code": code, "message": message}
self._response.status = "failed"
yield ErrorEvent(
type="response.error",
sequence_number=self.alloc_sequence_number(),
code=code,
message=message,
param=param,
)
yield ResponseEvent(
type="response.failed",
sequence_number=self.alloc_sequence_number(),
response=self._response,
)
Loading