fix(prompts): tighten prompt tuning output guidance - #136
Merged
Conversation
📝 WalkthroughWalkthroughThis PR introduces three independent feature updates: client-supplied request ID support for tracking user interactions, executor-based hard timeout enforcement for LLM calls to prevent indefinite blocking, and significant updates to playbook consolidation and extraction prompts with stricter output constraints and refined decision logic. ChangesClient Request ID Support
LLM Hard Timeout Protection
Prompt Library Updates
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (2)
tests/server/services/test_generation_service.py (1)
74-107: ⚡ Quick winConsider extracting common test setup to reduce duplication.
The test setup (lines 75-91) is nearly identical to
test_publish_request_with_session_id(lines 47-62). Extract the common setup into a fixture or helper function to improve maintainability:♻️ Example refactor using a fixture
+@pytest.fixture +def generation_service_setup(mock_llm_responses): + """Fixture providing configured generation service and sample interaction""" + org_id = "test_org" + with tempfile.TemporaryDirectory() as temp_dir: + llm_config = LiteLLMConfig(model="gpt-4o-mini") + llm_client = LiteLLMClient(llm_config) + generation_service = GenerationService( + llm_client=llm_client, + request_context=RequestContext(org_id=org_id, storage_base_dir=temp_dir), + ) + interaction = InteractionData( + content="test interaction", + created_at=int(datetime.datetime.now(UTC).timestamp()), + ) + yield generation_service, interaction + -def test_publish_request_honors_caller_request_id(mock_llm_responses): +def test_publish_request_honors_caller_request_id(generation_service_setup): - user_id = "test_user_id" - org_id = "test_org" + generation_service, interaction = generation_service_setup + user_id = "test_user_id" session_id = "test_session_id" request_id = "caller-request-id" - - with tempfile.TemporaryDirectory() as temp_dir: - llm_config = LiteLLMConfig(model="gpt-4o-mini") - llm_client = LiteLLMClient(llm_config) - generation_service = GenerationService( - llm_client=llm_client, - request_context=RequestContext(org_id=org_id, storage_base_dir=temp_dir), - ) - - interaction = InteractionData( - content="test interaction", - created_at=int(datetime.datetime.now(UTC).timestamp()), - ) - - request = PublishUserInteractionRequest( + + request = PublishUserInteractionRequest( + request_id=request_id, + user_id=user_id, + interaction_data_list=[interaction], + session_id=session_id, + )🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/server/services/test_generation_service.py` around lines 74 - 107, The two tests test_publish_request_honors_caller_request_id and test_publish_request_with_session_id duplicate setup for LiteLLMConfig/LiteLLMClient, RequestContext, GenerationService, and InteractionData; extract that repeated setup into a pytest fixture or helper function (e.g., a fixture named generation_service_factory or common_setup) that returns a configured GenerationService (or returns the tuple of generation_service, temp_dir, request_context) and reuse it in both tests; update both tests to call the fixture/helper and remove the duplicated lines creating llm_config, llm_client, RequestContext, GenerationService, and InteractionData while keeping assertions unchanged.reflexio/server/services/generation_service.py (1)
175-177: 💤 Low valueConsider explicit validation for empty string request_id.
The expression
publish_user_interaction_request.request_id or str(uuid.uuid4())treats an empty string""as falsy and silently replaces it with a UUID. While this is consistent with Python idioms, it may surprise callers who explicitly passrequest_id=""(which passes Pydantic validation since the field is typed asstr | None).Consider either:
- Adding a Pydantic validator to reject empty strings if they're invalid
- Explicitly checking
is Noneif you only want to replaceNone:if publish_user_interaction_request.request_id is None: ...This ensures the API contract is clear about what constitutes a "provided" request_id.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@reflexio/server/services/generation_service.py` around lines 175 - 177, The current line using publish_user_interaction_request.request_id or str(uuid.uuid4()) will treat an empty string as missing; change this to explicitly distinguish None from empty string: either (A) replace the expression with an explicit None check (e.g., if publish_user_interaction_request.request_id is None: request_id = str(uuid.uuid4()) else: request_id = publish_user_interaction_request.request_id) so empty strings are preserved, or (B) add a Pydantic validator on the request model that rejects empty strings for the request_id field (e.g., in the model used to construct publish_user_interaction_request) and then keep the current logic; update code in generation_service.py around the request_id assignment and the model class where request_id is defined accordingly.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@reflexio/server/llm/litellm_client.py`:
- Around line 980-983: The code directly calls
float(os.environ.get("REFLEXIO_LLM_HARD_TIMEOUT_GRACE_SECONDS", "5")) which will
raise on malformed env values; change this to parse with a safe try/except
(catch ValueError/TypeError), falling back to the default 5.0 if parsing fails,
then clamp with grace_seconds = max(0.0, parsed_value) before computing
hard_timeout = max(0.001, timeout_seconds) + grace_seconds; optionally emit a
warning/log when parsing fails. This affects the
REFLEXIO_LLM_HARD_TIMEOUT_GRACE_SECONDS env read and the grace_seconds /
hard_timeout calculation in litellm_client.py.
- Around line 985-997: The current _completion_with_hard_timeout implementation
spawns a ThreadPoolExecutor per request which cannot kill a blocking
litellm.completion call, allowing hung provider calls to accumulate stuck
threads; replace the thread-based approach with process-based isolation: run
litellm.completion in a separate process (e.g., via multiprocessing.Process or
concurrent.futures.ProcessPoolExecutor) and communicate the result back over a
Pipe/Queue, enforce the hard_timeout by terminating the child process if it
exceeds the timeout, raise LLMHardTimeoutError on termination, and ensure you
properly join/terminate the child and clean up resources (replace usages of
future.cancel() / executor.shutdown(...) in _completion_with_hard_timeout with
process termination and safe result collection).
In `@reflexio/server/services/generation_service.py`:
- Around line 175-177: The code currently coerces empty strings into new UUIDs
causing duplicate client request_ids to overwrite Request rows but still append
Interactions; change the assignment so a provided empty string is treated as a
supplied id (generate a uuid only when
publish_user_interaction_request.request_id is None), i.e. treat None vs ""
distinctly for publish_user_interaction_request.request_id; then in
GenerationService.run use RequestMixin.get_request(request_id) (or equivalent
existence check) before calling RequestMixin.add_request() and
add_user_interactions_bulk()/ _insert_interaction so you either return/no-op or
raise on a repeated request_id to enforce idempotency (or explicitly dedupe
interactions) and adjust
GenerationService.get_interaction_from_publish_user_interaction_request/_insert_interaction
logic as needed to avoid always appending when interaction_id == 0.
---
Nitpick comments:
In `@reflexio/server/services/generation_service.py`:
- Around line 175-177: The current line using
publish_user_interaction_request.request_id or str(uuid.uuid4()) will treat an
empty string as missing; change this to explicitly distinguish None from empty
string: either (A) replace the expression with an explicit None check (e.g., if
publish_user_interaction_request.request_id is None: request_id =
str(uuid.uuid4()) else: request_id =
publish_user_interaction_request.request_id) so empty strings are preserved, or
(B) add a Pydantic validator on the request model that rejects empty strings for
the request_id field (e.g., in the model used to construct
publish_user_interaction_request) and then keep the current logic; update code
in generation_service.py around the request_id assignment and the model class
where request_id is defined accordingly.
In `@tests/server/services/test_generation_service.py`:
- Around line 74-107: The two tests
test_publish_request_honors_caller_request_id and
test_publish_request_with_session_id duplicate setup for
LiteLLMConfig/LiteLLMClient, RequestContext, GenerationService, and
InteractionData; extract that repeated setup into a pytest fixture or helper
function (e.g., a fixture named generation_service_factory or common_setup) that
returns a configured GenerationService (or returns the tuple of
generation_service, temp_dir, request_context) and reuse it in both tests;
update both tests to call the fixture/helper and remove the duplicated lines
creating llm_config, llm_client, RequestContext, GenerationService, and
InteractionData while keeping assertions unchanged.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 6a9f911a-2e9c-4b83-bcde-d65bf7e8e712
📒 Files selected for processing (12)
reflexio/models/api_schema/domain/entities.pyreflexio/server/llm/litellm_client.pyreflexio/server/prompt/prompt_bank/playbook_consolidation/v2.3.0.prompt.mdreflexio/server/prompt/prompt_bank/playbook_consolidation/v2.3.1.prompt.mdreflexio/server/prompt/prompt_bank/playbook_consolidation/v2.3.2.prompt.mdreflexio/server/prompt/prompt_bank/playbook_extraction_context/v4.2.0.prompt.mdreflexio/server/prompt/prompt_bank/playbook_extraction_context/v4.2.2.prompt.mdreflexio/server/prompt/prompt_bank/playbook_extraction_context/v4.2.3.prompt.mdreflexio/server/services/generation_service.pytests/server/llm/test_litellm_client_unit.pytests/server/services/test_generation_service.pytests/server/services/test_prompt_model_mapping.py
Comment on lines
+980
to
+983
| grace_seconds = float( | ||
| os.environ.get("REFLEXIO_LLM_HARD_TIMEOUT_GRACE_SECONDS", "5") | ||
| ) | ||
| hard_timeout = max(0.001, timeout_seconds) + max(0.0, grace_seconds) |
There was a problem hiding this comment.
Handle malformed hard-timeout grace env safely.
REFLEXIO_LLM_HARD_TIMEOUT_GRACE_SECONDS is parsed with float(...) without a guard. A bad env value will make every LLM request fail before dispatch.
Proposed fix
- grace_seconds = float(
- os.environ.get("REFLEXIO_LLM_HARD_TIMEOUT_GRACE_SECONDS", "5")
- )
+ grace_raw = os.environ.get("REFLEXIO_LLM_HARD_TIMEOUT_GRACE_SECONDS", "5")
+ try:
+ grace_seconds = float(grace_raw)
+ except (TypeError, ValueError):
+ self.logger.warning(
+ "Invalid REFLEXIO_LLM_HARD_TIMEOUT_GRACE_SECONDS=%r; falling back to 5.0",
+ grace_raw,
+ )
+ grace_seconds = 5.0🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@reflexio/server/llm/litellm_client.py` around lines 980 - 983, The code
directly calls float(os.environ.get("REFLEXIO_LLM_HARD_TIMEOUT_GRACE_SECONDS",
"5")) which will raise on malformed env values; change this to parse with a safe
try/except (catch ValueError/TypeError), falling back to the default 5.0 if
parsing fails, then clamp with grace_seconds = max(0.0, parsed_value) before
computing hard_timeout = max(0.001, timeout_seconds) + grace_seconds; optionally
emit a warning/log when parsing fails. This affects the
REFLEXIO_LLM_HARD_TIMEOUT_GRACE_SECONDS env read and the grace_seconds /
hard_timeout calculation in litellm_client.py.
Collaborator
Author
There was a problem hiding this comment.
Fixed in 080b2a2. REFLEXIO_LLM_HARD_TIMEOUT_GRACE_SECONDS now parses through a guarded helper, falls back to 5.0 on malformed values, clamps at non-negative, and has a regression test for the invalid-env path.
Comment on lines
+985
to
+997
| executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="litellm") | ||
| future = executor.submit(litellm.completion, **params) | ||
| try: | ||
| return future.result(timeout=hard_timeout) | ||
| except FuturesTimeoutError as exc: | ||
| future.cancel() | ||
| raise LLMHardTimeoutError( | ||
| f"LLM request exceeded hard timeout of {hard_timeout:.3f}s " | ||
| f"(provider timeout={provider_timeout!r})" | ||
| ) from exc | ||
| finally: | ||
| executor.shutdown(wait=False, cancel_futures=True) | ||
|
|
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
python - <<'PY'
import time, threading
from concurrent.futures import ThreadPoolExecutor
started = threading.Event()
def slow():
started.set()
time.sleep(5)
ex = ThreadPoolExecutor(max_workers=1)
f = ex.submit(slow)
started.wait(timeout=1)
print("cancel_before_shutdown:", f.cancel()) # expected: False (already running)
ex.shutdown(wait=False, cancel_futures=True)
print("future_done_immediately:", f.done()) # expected: False right after shutdown
time.sleep(0.2)
print("future_still_running:", not f.done()) # expected: True (until sleep completes)
PYRepository: ReflexioAI/reflexio
Length of output: 151
🏁 Script executed:
#!/bin/bash
set -euo pipefail
repo_root="."
file="reflexio/server/llm/litellm_client.py"
# Print around snippet lines 950-1035 with line numbers
nl -ba "$file" | sed -n '930,1040p'Repository: ReflexioAI/reflexio
Length of output: 106
🏁 Script executed:
#!/bin/bash
set -euo pipefail
file="reflexio/server/llm/litellm_client.py"
nl -ba "$file" | sed -n '930,1040p'Repository: ReflexioAI/reflexio
Length of output: 106
🏁 Script executed:
#!/bin/bash
set -euo pipefail
file="reflexio/server/llm/litellm_client.py"
# Print lines 960-1025 with line numbers using awk (nl not available)
awk 'NR>=960 && NR<=1025 {printf "%d\t%s\n", NR, $0}' "$file"Repository: ReflexioAI/reflexio
Length of output: 3059
🏁 Script executed:
#!/bin/bash
python - <<'PY'
import time, threading
from concurrent.futures import ThreadPoolExecutor
started = threading.Event()
def slow():
started.set()
time.sleep(5)
ex = ThreadPoolExecutor(max_workers=1)
f = ex.submit(slow)
started.wait(timeout=1)
print("cancel_before_shutdown:", f.cancel()) # already running -> False
ex.shutdown(wait=False, cancel_futures=True)
print("future_done_immediately:", f.done()) # still not done
time.sleep(0.2)
print("future_still_running:", not f.done()) # thread continues running
PYRepository: ReflexioAI/reflexio
Length of output: 151
🏁 Script executed:
#!/bin/bash
python - <<'PY'
import time, threading
from concurrent.futures import ThreadPoolExecutor
started = threading.Event()
def slow():
started.set()
time.sleep(5)
ex = ThreadPoolExecutor(max_workers=1)
f = ex.submit(slow)
started.wait(timeout=1)
print("cancel_before_shutdown:", f.cancel()) # False (already running)
ex.shutdown(wait=False, cancel_futures=True)
print("future_done_immediately:", f.done()) # False
time.sleep(0.2)
print("future_still_running:", not f.done()) # True
PYRepository: ReflexioAI/reflexio
Length of output: 151
🏁 Script executed:
#!/bin/bash
python - <<'PY'
import time, threading
from concurrent.futures import ThreadPoolExecutor
started = threading.Event()
def slow():
started.set()
time.sleep(5)
ex = ThreadPoolExecutor(max_workers=1)
f = ex.submit(slow)
started.wait(timeout=1)
print("cancel_before_shutdown:", f.cancel()) # False (already running)
ex.shutdown(wait=False, cancel_futures=True)
print("future_done_immediately:", f.done()) # False
time.sleep(0.2)
print("future_still_running:", not f.done()) # True
PYRepository: ReflexioAI/reflexio
Length of output: 151
Hard-timeout can accumulate stuck worker threads on provider hangs
_completion_with_hard_timeout() spins up a new ThreadPoolExecutor(max_workers=1) per request and times out by catching FuturesTimeoutError, then calling future.cancel() and executor.shutdown(wait=False, cancel_futures=True). future.cancel() won’t stop an already-running blocking litellm.completion call, so a provider that hangs keeps its worker thread alive until the call returns—repeated incidents can accumulate stuck threads.
Use process-based isolation (run litellm.completion in a subprocess and terminate the process on timeout) or another mechanism that can actually kill the in-flight request.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@reflexio/server/llm/litellm_client.py` around lines 985 - 997, The current
_completion_with_hard_timeout implementation spawns a ThreadPoolExecutor per
request which cannot kill a blocking litellm.completion call, allowing hung
provider calls to accumulate stuck threads; replace the thread-based approach
with process-based isolation: run litellm.completion in a separate process
(e.g., via multiprocessing.Process or concurrent.futures.ProcessPoolExecutor)
and communicate the result back over a Pipe/Queue, enforce the hard_timeout by
terminating the child process if it exceeds the timeout, raise
LLMHardTimeoutError on termination, and ensure you properly join/terminate the
child and clean up resources (replace usages of future.cancel() /
executor.shutdown(...) in _completion_with_hard_timeout with process termination
and safe result collection).
Collaborator
Author
There was a problem hiding this comment.
Fixed in 080b2a2. Real LiteLLM completions now run in a child process for the hard-timeout path; on timeout the process is terminated, then killed if needed, before raising LLMHardTimeoutError. The old per-request ThreadPoolExecutor path was removed.
Comment on lines
+175
to
+177
| request_id = publish_user_interaction_request.request_id or str( | ||
| uuid.uuid4() | ||
| ) |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Description: Search for existing tests or code that exercises duplicate request_id scenarios
# Check for tests that publish multiple times with the same request_id
rg -n -A5 -B5 'request_id.*=.*["\'].*["\']' tests/ | rg -A10 -B10 'PublishUserInteractionRequest'
# Check for any uniqueness constraints or validation on request_id in storage layer
rg -n 'UNIQUE.*request_id|request_id.*UNIQUE' reflexio/server/services/storage/
# Look for idempotency handling or duplicate detection
ast-grep --pattern 'if $_.request_id $$$'Repository: ReflexioAI/reflexio
Length of output: 132
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== Locate generation_service snippet =="
rg -n "request_id\s*=\s*publish_user_interaction_request\.request_id|publish_user_interaction_request\.request_id\s*or" -S reflexio/server/services/generation_service.py || true
echo
echo "== Inspect surrounding lines in generation_service.py (160-230) =="
nl -ba reflexio/server/services/generation_service.py | sed -n '150,240p'
echo
echo "== Find storage requests insert behavior =="
# locate files mentioning INSERT OR REPLACE and requests table
rg -n --hidden -S "INSERT\s+OR\s+REPLACE|insert\s+or\s+replace" reflexio/server/services/storage/ || true
rg -n --hidden -S "requests" reflexio/server/services/storage/ || true
echo
echo "== Find UNIQUE constraints or indexes for request_id in storage =="
rg -n --hidden -S "UNIQUE.*request_id|request_id.*UNIQUE|CREATE\s+UNIQUE|CREATE\s+INDEX.*request_id" reflexio/server/services/storage/ || true
echo
echo "== Search for PublishUserInteractionRequest usage in tests (duplicate scenarios) =="
rg -n "PublishUserInteractionRequest" tests/ || true
echo
echo "== Search for publish calls with explicit request_id in code/tests =="
rg -n -S "request_id\s*=\s*['\"][^'\"]+['\"]" tests/ reflexio/ || true
echo
echo "== Search for idempotency/duplicate handling keywords around request_id =="
rg -n -S "idempotent|idempotency|duplicate|conflict|already exists|REPLACE|ON CONFLICT" reflexio/server/services/ || true
echo
echo "== Check interactions table relation to request_id (foreign key / accumulation) =="
rg -n -S "CREATE\s+TABLE.*interactions|interactions.*request_id|request_id.*interactions" reflexio/server/services/storage/ || trueRepository: ReflexioAI/reflexio
Length of output: 313
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== Basic sanity: show generation_service request_id assignment =="
sed -n '150,210p' reflexio/server/services/generation_service.py
echo
echo "== Dump exact lines around request_id assignment (170-190) with line numbers via awk =="
awk 'NR>=165 && NR<=195 {printf "%6d:%s\n", NR, $0}' reflexio/server/services/generation_service.py
echo
echo "== Locate sqlite storage requests implementation (_requests.py) =="
fd -a "_requests.py" reflexio/server/services/storage || true
fd -a "sqlite_storage" reflexio/server/services/storage || true
echo
echo "== Search storage for INSERT OR REPLACE and requests table =="
rg -n --hidden -S "INSERT\s+OR\s+REPLACE|insert\s+or\s+replace" reflexio/server/services/storage/ || true
rg -n --hidden -S "CREATE\s+TABLE.*requests|requests" reflexio/server/services/storage/ || true
echo
echo "== Search for UNIQUE constraints / indexes for request_id =="
rg -n --hidden -S "UNIQUE.*request_id|request_id.*UNIQUE|CREATE\s+UNIQUE|CREATE\s+INDEX.*request_id" reflexio/server/services/storage/ || true
echo
echo "== Find where Requests are written (look for add_request / insert into requests) =="
rg -n --hidden -S "add_request|insert.*into.*requests|upsert|ON CONFLICT" reflexio/server/services/storage/ || true
echo
echo "== Inspect sqlite_storage/_requests.py if present =="
REQ_FILE="$(fd -a "_requests.py" reflexio/server/services/storage | head -n 1 || true)"
if [ -n "${REQ_FILE:-}" ]; then
echo "Using: $REQ_FILE"
sed -n '1,220p' "$REQ_FILE"
fi
echo
echo "== Search for tests exercising duplicate request_id / retries =="
rg -n -S "PublishUserInteractionRequest|request_id\s*=" tests/ reflexio/ || true
echo
echo "== Search for idempotency/duplicate handling keywords near request_id =="
rg -n -S "idempotent|idempotency|duplicate|conflict|already exists|REPLACE|ON CONFLICT" reflexio/server/services/ || true
echo
echo "== Inspect interactions schema relation to request_id =="
rg -n --hidden -S "CREATE\s+TABLE.*interactions|interactions.*request_id|request_id.*interactions|FOREIGN\s+KEY.*request_id" reflexio/server/services/storage/ || trueRepository: ReflexioAI/reflexio
Length of output: 50376
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== generation_service: show Request construction + persistence (around new_request / add_request) =="
rg -n "new_request\s*=\s*Request|add_request\(" reflexio/server/services/generation_service.py
sed -n '200,270p' reflexio/server/services/generation_service.py
echo
echo "== sqlite_storage schema: show CREATE TABLE requests + interactions =="
rg -n "CREATE TABLE IF NOT EXISTS requests|CREATE TABLE IF NOT EXISTS interactions" reflexio/server/services/storage/sqlite_storage/_base.py
sed -n '1580,1685p' reflexio/server/services/storage/sqlite_storage/_base.py
echo
echo "== storage insert path for interactions: look for INSERT/OR REPLACE into interactions =="
rg -n --hidden -S "INSERT\s+OR\s+REPLACE\s+INTO\s+interactions|INSERT\s+INTO\s+interactions|add_interaction|add_interactions" reflexio/server/services/storage/sqlite_storage | head -n 80
echo
echo "== Inspect RequestMixin + any interaction mixin file for add_interaction(s) =="
fd -a "*interaction*" reflexio/server/services/storage/sqlite_storage || true
fd -a "_interactions.py" reflexio/server/services/storage/sqlite_storage || true
fd -a "_interactions" reflexio/server/services/storage/sqlite_storage || true
# Try common filenames
for f in reflexio/server/services/storage/sqlite_storage/_interactions.py \
reflexio/server/services/storage/sqlite_storage/_interactions_mixin.py \
reflexio/server/services/storage/storage_base/_interactions.py; do
if [ -f "$f" ]; then
echo "== file exists: $f =="
sed -n '1,220p' "$f"
fi
done
echo
echo "== Search tests for behavior when same request_id is inserted twice =="
# Target the storage contract tests first (most likely to cover upserts/replaces)
rg -n "test_.*request_id|add_request|INSERT OR REPLACE|OR REPLACE INTO requests|requests.*request_id" tests/server/services/storage/test_storage_contract_requests.py tests/server/services/storage/test_sqlite_storage.py tests/server/services/storage/test_sqlite_storage_bc_extras.py || true
sed -n '1,220p' tests/server/services/storage/test_storage_contract_requests.py || true
echo
echo "== Check generation_service tests for publishing same request_id twice =="
sed -n '1,240p' tests/server/services/test_generation_service.py || trueRepository: ReflexioAI/reflexio
Length of output: 20387
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== Find add_user_interactions_bulk implementation =="
rg -n --hidden -S "def add_user_interactions_bulk|add_user_interactions_bulk\(" reflexio/server/services/storage || true
for f in $(rg -l --hidden -S "add_user_interactions_bulk" reflexio/server/services/storage | head -n 20); do
echo
echo "---- $f ----"
rg -n "add_user_interactions_bulk" "$f" -n
done
# Show relevant portions from the likely file
target_files=$(rg -l --hidden -S "def add_user_interactions_bulk" reflexio/server/services/storage | head -n 5 || true)
if [ -n "${target_files:-}" ]; then
for f in $target_files; do
echo
echo "== Showing function area in $f =="
# print around first match line
line=$(rg -n "def add_user_interactions_bulk" "$f" | head -n 1 | cut -d: -f1)
start=$((line-40)); end=$((line+140))
sed -n "${start},${end}p" "$f"
done
fi
echo
echo "== Confirm interactions INSERT/UPSERT behavior =="
rg -n --hidden -S "INSERT\s+INTO\s+interactions|INSERT\s+OR\s+REPLACE\s+INTO\s+interactions|ON\s+CONFLICT.*interactions" reflexio/server/services/storage/sqlite_storage || true
echo
echo "== Inspect publish tests for any duplicate request_id publishing =="
sed -n '1,260p' tests/server/services/test_generation_service.py
echo
echo "== Continue inspect further in test_generation_service.py (260-520) =="
sed -n '260,560p' tests/server/services/test_generation_service.py
echo
echo "== Search within test_generation_service.py for 'publish' calls with same request_id var =="
rg -n "request_id\s*=" tests/server/services/test_generation_service.py
rg -n -S "caller-request-id|test_request|req1|r1" tests/server/services/test_generation_service.py || trueRepository: ReflexioAI/reflexio
Length of output: 20456
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== Show _insert_interaction / add_interaction section around INSERT OR REPLACE INTO interactions (sqlite_storage/_profiles.py:360-430) =="
sed -n '330,460p' reflexio/server/services/storage/sqlite_storage/_profiles.py
echo
echo "== Search generation_service.py for any existing-request/idempotency check (get_request/duplicate handling) =="
rg -n --hidden -S "get_request\(|request_id.*get|existing.*request|duplicate|idempotent|already.*request" reflexio/server/services/generation_service.py || true
echo
echo "== Search for any early return when request_id exists =="
rg -n --hidden -S "if .*request_id.*(exists|present|get_request)" reflexio/server/services/generation_service.py || true
echo
echo "== Inspect PublishUserInteractionRequest schema for request_id constraints/typing (if in models) =="
# find definition
REQ_FILE="$(fd -a "service_schemas.py" reflexio/models/api_schema/service_schemas | head -n 1 || true)"
echo "service_schemas file: $REQ_FILE"
# If not found, just search for class name
rg -n "class\s+PublishUserInteractionRequest|PublishUserInteractionRequest" reflexio/models -S | head -n 20Repository: ReflexioAI/reflexio
Length of output: 6473
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== Show _insert_interaction / add_interaction section around INSERT OR REPLACE INTO interactions (sqlite_storage/_profiles.py:360-460) =="
sed -n '330,460p' reflexio/server/services/storage/sqlite_storage/_profiles.py
echo
echo "== Search generation_service.py for any existing-request/idempotency check (get_request/duplicate handling) =="
rg -n --hidden -S "get_request\(|request_id.*get|existing.*request|duplicate|idempotent|already.*request" reflexio/server/services/generation_service.py || true
echo
echo "== Search for any early return when request_id exists =="
rg -n --hidden -S "if .*request_id.*(exists|present|get_request)" reflexio/server/services/generation_service.py || true
echo
echo "== Inspect PublishUserInteractionRequest schema for request_id constraints/typing =="
rg -n "PublishUserInteractionRequest" reflexio/models/api_schema -S | head -n 50Repository: ReflexioAI/reflexio
Length of output: 6287
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== Show _insert_interaction / add_interaction section around INSERT OR REPLACE INTO interactions (sqlite_storage/_profiles.py:330-460) =="
sed -n '330,460p' reflexio/server/services/storage/sqlite_storage/_profiles.py
echo
echo "== Search generation_service.py for any existing-request/idempotency check (get_request/duplicate handling) =="
rg -n --hidden -S "get_request\(|existing.*request|duplicate|idempotent|already.*request" reflexio/server/services/generation_service.py || true
echo
echo "== Inspect PublishUserInteractionRequest schema for request_id constraints/typing =="
rg -n "class\s+PublishUserInteractionRequest|PublishUserInteractionRequest" reflexio/models/api_schema -S | head -n 80Repository: ReflexioAI/reflexio
Length of output: 6229
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== Inspect PublishUserInteractionRequest.request_id field/validators (entities.py around class) =="
sed -n '530,640p' reflexio/models/api_schema/domain/entities.py
echo
echo "== Search for any validators/precondition checks referencing request_id =="
rg -n --hidden -S "request_id.*(min_length|max_length|validator|field_validator|model_validator|constr|required)|validate_.*request_id|duplicate.*request_id|idempot" reflexio/models reflexio/server tests | head -n 200
echo
echo "== Inspect Request model definition (created_at default) =="
# Locate class Request in entities.py or service_schemas
rg -n "class\s+Request\b" reflexio/models/api_schema -S
rg -n "Request\(" reflexio/models/api_schema -S | head -n 50
# Print the Request model area near where it's defined
REQ_LOC="$(rg -n "class\s+Request\b" reflexio/models/api_schema -S | head -n 1 | cut -d: -f1 || true)"
if [ -n "${REQ_LOC:-}" ]; then
start=$((REQ_LOC-40)); end=$((REQ_LOC+120))
sed -n "${start},${end}p" "$(echo reflexio/models/api_schema/domain/entities.py)" || true
fi
# Also try if Request is in a different file
for f in $(fd -e py "service_schemas.py" reflexio/models | head -n 5); do
echo "== file: $f =="
rg -n "class\s+Request\b" "$f" -S || true
doneRepository: ReflexioAI/reflexio
Length of output: 13903
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== Request model (entities.py) around class Request =="
# entities.py is the one we saw earlier
sed -n '160,260p' reflexio/models/api_schema/domain/entities.py
echo
echo "== PublishUserInteractionRequest.request_id usage in Request construction =="
# show relevant generation_service block already partially shown, but include Request init region
sed -n '190,240p' reflexio/server/services/generation_service.py
echo
echo "== GenerationService.get_interaction_from_publish_user_interaction_request =="
# locate method
rg -n "get_interaction_from_publish_user_interaction_request" reflexio/server/services/generation_service.py
method_line=$(rg -n "get_interaction_from_publish_user_interaction_request" reflexio/server/services/generation_service.py | head -n 1 | cut -d: -f1)
start=$((method_line-20)); end=$((method_line+140))
sed -n "${start},${end}p" reflexio/server/services/generation_service.py
echo
echo "== Any early return / idempotency gate in GenerationService.run =="
rg -n --hidden -S "if .*request_id|request_id.*(get_request|exists|seen|duplicate|idempot)" reflexio/server/services/generation_service.py
echo
echo "== Operation queue idempotency logic (storage_base/_operations.py) =="
sed -n '120,220p' reflexio/server/services/storage/storage_base/_operations.py
echo
echo "== Tests for queue ignoring duplicate request_id (storage/test_storage_contract_operations.py around 127) =="
sed -n '90,170p' tests/server/services/storage/test_storage_contract_operations.pyRepository: ReflexioAI/reflexio
Length of output: 14653
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== Interaction model definition (entities.py) around class Interaction =="
# Find Interaction class
I_LINE=$(rg -n "class\s+Interaction\b" reflexio/models/api_schema/domain/entities.py | head -n 1 | cut -d: -f1)
if [ -z "${I_LINE:-}" ]; then
rg -n "Interaction\s*=\s*BaseModel" reflexio/models/api_schema/domain/entities.py || true
else
start=$((I_LINE-40)); end=$((I_LINE+120))
sed -n "${start},${end}p" reflexio/models/api_schema/domain/entities.py
fi
echo
echo "== GenerationService.get_interaction_from_publish_user_interaction_request implementation =="
M_LINE=$(rg -n "def\s+get_interaction_from_publish_user_interaction_request" reflexio/server/services/generation_service.py | head -n 1 | cut -d: -f1)
start=$((M_LINE-20)); end=$((M_LINE+220))
sed -n "${start},${end}p" reflexio/server/services/generation_service.py
echo
echo "== storage_base/_operations.py: look for duplicate request_id/idempotent queue logic =="
sed -n '120,200p' reflexio/server/services/storage/storage_base/_operations.py
echo
echo "== tests/server/services/storage/test_storage_contract_requests.py: check duplicate request_id add_request usage =="
sed -n '1,120p' tests/server/services/storage/test_storage_contract_requests.py
echo
echo "== tests/server/services/storage/test_storage_contract_operations.py around duplicate request_id queue ignore =="
sed -n '90,170p' tests/server/services/storage/test_storage_contract_operations.pyRepository: ReflexioAI/reflexio
Length of output: 19647
Fix non-idempotent request_id retries: duplicates overwrite requests but still insert new interactions
In reflexio/server/services/generation_service.py the effective id is computed as:
request_id = publish_user_interaction_request.request_id or str(uuid.uuid4())- Empty string is treated as falsy, so
""becomes a new UUID (sincePublishUserInteractionRequest.request_idisstr | Nonewith no non-empty constraint). - The SQLite
requeststable usesrequest_id TEXT PRIMARY KEY, andRequestMixin.add_request()persists viaINSERT OR REPLACE INTO requests, so a duplicate clientrequest_idoverwrites the priorRequestrow (changingcreated_at/metadata/etc.). - Each publish with that same
request_idstill inserts newinteractions:GenerationService.get_interaction_from_publish_user_interaction_request()leavesInteraction.interaction_idat the default0, and_insert_interaction()usesif interaction.interaction_id:to decide betweenINSERT OR REPLACE(only wheninteraction_idis non-zero) vsINSERT INTO interactions(append). GenerationService.run()always performsadd_request(...)+add_user_interactions_bulk(...)for the computedrequest_id(no early return when therequest_idalready exists). The only dedupe found is for the operation pending queue while blocked (try_acquire_in_progress_lockdrops duplicaterequest_identries), which doesn’t make the whole publish flow idempotent.
Either enforce idempotency semantics for client-supplied request_id (validate non-empty + return/no-op or otherwise prevent re-processing/interaction duplication) or explicitly document the “overwrite requests + append interactions” behavior and ensure downstream logic can’t be inconsistent.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@reflexio/server/services/generation_service.py` around lines 175 - 177, The
code currently coerces empty strings into new UUIDs causing duplicate client
request_ids to overwrite Request rows but still append Interactions; change the
assignment so a provided empty string is treated as a supplied id (generate a
uuid only when publish_user_interaction_request.request_id is None), i.e. treat
None vs "" distinctly for publish_user_interaction_request.request_id; then in
GenerationService.run use RequestMixin.get_request(request_id) (or equivalent
existence check) before calling RequestMixin.add_request() and
add_user_interactions_bulk()/ _insert_interaction so you either return/no-op or
raise on a repeated request_id to enforce idempotency (or explicitly dedupe
interactions) and adjust
GenerationService.get_interaction_from_publish_user_interaction_request/_insert_interaction
logic as needed to avoid always appending when interaction_id == 0.
Collaborator
Author
There was a problem hiding this comment.
Fixed in 080b2a2. request_id is now NonEmptyStr | None, UUID generation only happens for None, and duplicate caller-supplied request IDs are rejected before Request or Interaction writes. Added tests for empty IDs and duplicate publish attempts.
yilu331
added a commit
that referenced
this pull request
Jun 9, 2026
## Summary - Mark active extraction agent runs failed when extractor service timeout handling fires for a request. - Guard resumable extraction completion updates with expected active statuses so late LLM/tool output cannot commit after a run is already failed. - Add SQLite storage and resumable-agent regression coverage for status-guarded updates and late-output discard. ## Context This is the focused replacement for the useful remaining backend behavior from the now-closed #131. Prompt tuning, request-id, and hard-timeout work already landed separately in #136. ## Tests - `uv run python -c "import reflexio"` - `uv run ruff check reflexio/server/services/base_generation_service.py reflexio/server/services/extraction/resumable_agent.py reflexio/server/services/storage/sqlite_storage/_agent_run.py reflexio/server/services/storage/storage_base/_agent_run.py tests/server/services/extraction/test_resumable_agent.py tests/server/services/storage/sqlite_storage/test_agent_run_storage.py` - `uv run pyright reflexio/server/services/base_generation_service.py reflexio/server/services/extraction/resumable_agent.py reflexio/server/services/storage/sqlite_storage/_agent_run.py reflexio/server/services/storage/storage_base/_agent_run.py tests/server/services/extraction/test_resumable_agent.py tests/server/services/storage/sqlite_storage/test_agent_run_storage.py` - `uv run pytest tests/server/services/extraction/test_resumable_agent.py tests/server/services/storage/sqlite_storage/test_agent_run_storage.py -q --no-cov` <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit ## Bug Fixes * Enhanced extraction timeout management to properly mark failed extraction runs in the system * Prevents processing and committing outputs that arrive after an extraction operation has timed out and been marked as failed ## Tests * Added tests to verify extraction timeout handling and validation of late-output rejection behavior <!-- end of auto-generated comment: release notes by coderabbit.ai -->
3 tasks
This was referenced Jul 27, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Verification
Summary by CodeRabbit
New Features
Updates