feat(sse) : added sse support so frontend can see the status of agent… - #16
Merged
Conversation
📝 WalkthroughWalkthroughThe change adds Redis-backed investigation pipeline events and SSE streaming, updates incident and sampling lifecycle handling, propagates prompts through evaluation, and constrains issue-agent repository exploration with tool budgets, retries, and caches. ChangesIncident pipeline and investigation flow
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant TraceIngestion
participant PipelinePublisher
participant Redis
participant InvestigationStream
participant Client
TraceIngestion->>PipelinePublisher: publish pipeline stages
PipelinePublisher->>Redis: publish incident event
InvestigationStream->>Redis: subscribe to incident channel
Redis-->>InvestigationStream: deliver PipelineEvent
InvestigationStream-->>Client: send SSE event
Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.github.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 5
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
apps/node-api/src/worker.ts (1)
246-285: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winPublish the remaining issue-agent sub-steps from the Python worker The Node side only emits
BUILDING_TIMELINEandSEARCHING_REPOSITORY, then blocks onrunIssueAgent(...); the Python agent does not emit any later sub-step events. That leaves the SSE stream stuck on “Searching repository” for the rest of the investigation.🤖 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 `@apps/node-api/src/worker.ts` around lines 246 - 285, The issue-agent flow around runIssueAgent currently publishes only SEARCHING_REPOSITORY, leaving subsequent investigation progress absent from the SSE stream. Update the Node/Python issue-agent integration to emit the remaining sub-step events produced by the Python worker after repository searching, and ensure those events are forwarded through pipeline.subStep so progress advances while runIssueAgent executes.
🧹 Nitpick comments (7)
packages/issue-agent/src/issue_agent/agent.py (3)
141-142: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueReturn a JSON-shaped budget warning here too.
search_repoandbuild_code_graphreturn JSON objects on budget exhaustion;read_filereturns bare prose. Keeping one shape makes the signal easier for the model (and any log parsing) to recognize.🤖 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 `@packages/issue-agent/src/issue_agent/agent.py` around lines 141 - 142, Update the budget-exhaustion return in the agent flow around ctx.deps.tool_calls_used and MAX_TOOL_CALL_BUDGET to return a JSON-shaped warning consistent with search_repo and build_code_graph, rather than bare prose. Preserve the existing message content and report-completion guidance within the structured response.
105-107: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDerive the budget number from
MAX_TOOL_CALL_BUDGETinstead of hardcoding "8".The prompt and the enforcement constant will drift the moment the budget changes.
♻️ Proposed refactor
-## Rules & Constraints -- **Strict Budget**: At most 8 tool calls total. Focus on high-relevance searches and 1-3 key files. +## Rules & Constraints +- **Strict Budget**: At most {max_tool_calls} tool calls total. Focus on high-relevance searches and 1-3 key files.Then format once at module scope:
SYSTEM_PROMPT = _SYSTEM_PROMPT_TEMPLATE.format(max_tool_calls=MAX_TOOL_CALL_BUDGET)🤖 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 `@packages/issue-agent/src/issue_agent/agent.py` around lines 105 - 107, Update the prompt construction in the module containing the Rules & Constraints text to use a template placeholder for the budget, then format it once at module scope using MAX_TOOL_CALL_BUDGET. Remove the hardcoded “8” while preserving the existing prompt content and enforcement behavior.
126-162: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winConsider not charging budget for duplicate calls now that the repo layer caches.
repository.pynow serves repeatedsearch_symbol/read_file/build_code_graphcalls from memory, but each repeat still consumes one of the 8 slots. Since dedup is only enforced by prompt text, a model that repeats itself loses real investigation depth for zero external cost. Skipping the increment on a cache hit (or on an already-seen query/path) makes the budget track actual work.♻️ Sketch for `search_repo` / `read_file`
def search_repo(ctx: RunContext[IssueAgentDeps], query: str) -> str: """Search repository files for a symbol, function, or keyword.""" - if ctx.deps.tool_calls_used >= MAX_TOOL_CALL_BUDGET: + is_repeat = query in ctx.deps.symbols_searched + if not is_repeat and ctx.deps.tool_calls_used >= MAX_TOOL_CALL_BUDGET: return json.dumps({"warning": "Tool budget limit reached. Summarize findings and complete the report."}) logger.info("[repo] search_repo query=%r", query) result = ctx.deps.repo.search_symbol(query) - if query not in ctx.deps.symbols_searched: + if not is_repeat: ctx.deps.symbols_searched.append(query) - ctx.deps.tool_calls_used += 1 + ctx.deps.tool_calls_used += 1 return json.dumps(result, indent=2)🤖 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 `@packages/issue-agent/src/issue_agent/agent.py` around lines 126 - 162, Update search_repo, read_file, and build_code_graph so repeated cached calls do not increment ctx.deps.tool_calls_used. Detect duplicates using the existing symbols_searched/files_read tracking or the repository cache-hit result, while preserving the current budget-limit checks and recording first-time searches/reads as before.apps/node-api/src/pipeline.ts (2)
32-34: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winNo
errorlistener on either Redis client.Neither
createPipelinePublishernorcreatePipelineSubscriberattach an.on('error', ...)handler. Modern ioredis won't crash the process for this, but connection failures will only surface as a generic[ioredis] Unhandled error eventconsole line with no application-level visibility (no reconnect logging, no metrics).🩺 Suggested fix
export function createPipelinePublisher(redisUrl: string) { const publisher = new IORedis(redisUrl); + publisher.on('error', (err) => console.error('[pipeline] publisher connection error:', err));Also applies to: 71-73
🤖 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 `@apps/node-api/src/pipeline.ts` around lines 32 - 34, Add application-level error listeners to both Redis clients created by createPipelinePublisher and createPipelineSubscriber, logging each connection error through the existing application logging mechanism. Ensure both publisher and subscriber errors are handled consistently without changing their publishing or subscription behavior.
71-98: 🚀 Performance & Scalability | 🔵 Trivial
createPipelineSubscriberopens a brand-newIORedisconnection per call and adds a newon('message', ...)listener on everysubscribe()without removing it inunsubscribe().index.ts's SSE route callscreatePipelineSubscriber(REDIS_URL)fresh for every client connection, so each concurrent viewer opens a dedicated Redis connection instead of sharing one multiplexed subscription.
apps/node-api/src/pipeline.ts#L71-L98: refactor to a single shared subscriber that multiplexes channels via a registry (channel → callback), removing listeners onunsubscribe().apps/node-api/src/index.ts#L146-L149: use one long-lived shared subscriber instance (e.g., app-level singleton) instead of instantiatingcreatePipelineSubscriberper SSE request.
[medium]🤖 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 `@apps/node-api/src/pipeline.ts` around lines 71 - 98, Refactor createPipelineSubscriber in apps/node-api/src/pipeline.ts (lines 71-98) to maintain one IORedis connection and a channel-to-callback registry, dispatching messages through a shared listener and removing registry entries and channel subscriptions in unsubscribe. Update the SSE route in apps/node-api/src/index.ts (lines 146-149) to reuse one long-lived createPipelineSubscriber instance rather than creating one per client connection.apps/node-api/src/index.ts (1)
354-357: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winSequential awaited Redis publishes on the trace-ingestion hot path.
Three
await pipeline.stage(...)calls now block the/api/tracesresponse on Redis round-trips. Sincepipeline.publish()already swallows errors internally, these are best-effort observability writes and don't need to gate the response.⚡ Suggested fix
- await pipeline.stage(incidentId, 'RISK_ENGINE', 'completed', `Severity: ${risk.severity}`); - await pipeline.stage(incidentId, 'INCIDENT_CREATED', 'completed', `${result.action === 'CREATED' ? 'New' : 'Updated'} incident`); - await pipeline.stage(incidentId, 'EVALUATOR', 'running', 'LLM evaluator queued'); + void pipeline.stage(incidentId, 'RISK_ENGINE', 'completed', `Severity: ${risk.severity}`); + void pipeline.stage(incidentId, 'INCIDENT_CREATED', 'completed', `${result.action === 'CREATED' ? 'New' : 'Updated'} incident`); + void pipeline.stage(incidentId, 'EVALUATOR', 'running', 'LLM evaluator queued');🤖 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 `@apps/node-api/src/index.ts` around lines 354 - 357, Update the three pipeline.stage calls in the trace-ingestion flow to publish best-effort stages without awaiting each Redis round-trip, allowing the /api/traces response path to continue immediately. Preserve the existing incidentId, stage names, statuses, and messages, and rely on pipeline.publish()’s existing error handling.apps/node-api/src/worker.ts (1)
137-152: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDetailed evaluator error message is discarded before persistence.
errMsgis logged to console but not stored;engineering_reportis overwritten with just{ error: { code: 'EVALUATOR_FAILED' } }. PersistingerrMsgalongside the code would aid later debugging without requiring log access.🩹 Suggested fix
- engineering_report: { error: { code: 'EVALUATOR_FAILED' } } as object, + engineering_report: { error: { code: 'EVALUATOR_FAILED', message: errMsg } } as object,🤖 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 `@apps/node-api/src/worker.ts` around lines 137 - 152, Update the evaluator error handling in the catch block to persist the existing errMsg in engineering_report.error alongside the EVALUATOR_FAILED code. Keep the current console logging, pipeline failure stage, and analysis_status update unchanged.
🤖 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 `@apps/node-api/src/index.ts`:
- Around line 223-237: Add authentication and authorization to the POST
/api/admin/reset/queue handler before creating the Redis client or flushing
data, and replace redis.flushall() with targeted removal of only the intended
queue keys. Update the misleading queue-only comment and preserve the existing
success/error responses.
- Around line 109-160: Update the `/api/investigations/:incidentId/stream`
handler to validate `incident` immediately after the database lookup and reject
nonexistent incidents before opening the SSE stream or Redis subscription.
Establish the `createPipelineSubscriber` subscription before emitting any
initial synthetic stages, while preserving the existing initial-state payloads
and cleanup behavior so events published during initialization are not missed.
In `@packages/evaluator/src/evaluator/prompt_builder.py`:
- Around line 115-116: Update the prompt-building logic around context.prompt to
treat the user request as untrusted evidence: truncate it to a defined maximum
length, then append it with clear delimiters and an explicit untrusted-content
label. Preserve the existing omission behavior when context.prompt is empty.
In `@packages/issue-agent/src/issue_agent/repository.py`:
- Around line 131-134: Update GitHubRepo._safe_get to preserve the HTTP status
code for failures, then in the file-loading cache path at
packages/issue-agent/src/issue_agent/repository.py:131-134 cache None only for
genuine 404 responses; transient failures must not be cached. In the search
caching path at packages/issue-agent/src/issue_agent/repository.py:117-124, skip
self._search_cache[symbol] = result when both content_data and tree_data contain
"error", allowing later attempts to recover.
- Around line 229-235: Update the exception handling around the read_text call
in the repository file-reading method to catch OSError alongside ValueError,
allowing filesystem failures such as PermissionError to follow the existing
None-cache and return path. Simplify the exception tuple since
UnicodeDecodeError is already covered by ValueError.
---
Outside diff comments:
In `@apps/node-api/src/worker.ts`:
- Around line 246-285: The issue-agent flow around runIssueAgent currently
publishes only SEARCHING_REPOSITORY, leaving subsequent investigation progress
absent from the SSE stream. Update the Node/Python issue-agent integration to
emit the remaining sub-step events produced by the Python worker after
repository searching, and ensure those events are forwarded through
pipeline.subStep so progress advances while runIssueAgent executes.
---
Nitpick comments:
In `@apps/node-api/src/index.ts`:
- Around line 354-357: Update the three pipeline.stage calls in the
trace-ingestion flow to publish best-effort stages without awaiting each Redis
round-trip, allowing the /api/traces response path to continue immediately.
Preserve the existing incidentId, stage names, statuses, and messages, and rely
on pipeline.publish()’s existing error handling.
In `@apps/node-api/src/pipeline.ts`:
- Around line 32-34: Add application-level error listeners to both Redis clients
created by createPipelinePublisher and createPipelineSubscriber, logging each
connection error through the existing application logging mechanism. Ensure both
publisher and subscriber errors are handled consistently without changing their
publishing or subscription behavior.
- Around line 71-98: Refactor createPipelineSubscriber in
apps/node-api/src/pipeline.ts (lines 71-98) to maintain one IORedis connection
and a channel-to-callback registry, dispatching messages through a shared
listener and removing registry entries and channel subscriptions in unsubscribe.
Update the SSE route in apps/node-api/src/index.ts (lines 146-149) to reuse one
long-lived createPipelineSubscriber instance rather than creating one per client
connection.
In `@apps/node-api/src/worker.ts`:
- Around line 137-152: Update the evaluator error handling in the catch block to
persist the existing errMsg in engineering_report.error alongside the
EVALUATOR_FAILED code. Keep the current console logging, pipeline failure stage,
and analysis_status update unchanged.
In `@packages/issue-agent/src/issue_agent/agent.py`:
- Around line 141-142: Update the budget-exhaustion return in the agent flow
around ctx.deps.tool_calls_used and MAX_TOOL_CALL_BUDGET to return a JSON-shaped
warning consistent with search_repo and build_code_graph, rather than bare
prose. Preserve the existing message content and report-completion guidance
within the structured response.
- Around line 105-107: Update the prompt construction in the module containing
the Rules & Constraints text to use a template placeholder for the budget, then
format it once at module scope using MAX_TOOL_CALL_BUDGET. Remove the hardcoded
“8” while preserving the existing prompt content and enforcement behavior.
- Around line 126-162: Update search_repo, read_file, and build_code_graph so
repeated cached calls do not increment ctx.deps.tool_calls_used. Detect
duplicates using the existing symbols_searched/files_read tracking or the
repository cache-hit result, while preserving the current budget-limit checks
and recording first-time searches/reads as before.
🪄 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: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 0aba87ed-b893-4c50-94d4-bdaa199150ad
📒 Files selected for processing (12)
apps/node-api/src/index.tsapps/node-api/src/pipeline.tsapps/node-api/src/sampling-consumer.tsapps/node-api/src/worker.tspackages/adaptive-sampling/src/service.tspackages/evaluator/src/evaluator/context_builder.pypackages/evaluator/src/evaluator/prompt_builder.pypackages/evaluator/src/evaluator/schemas.pypackages/incident-formation/src/service.tspackages/issue-agent/src/issue_agent/__main__.pypackages/issue-agent/src/issue_agent/agent.pypackages/issue-agent/src/issue_agent/repository.py
Comment on lines
+109
to
+160
| app.get('/api/investigations/:incidentId/stream', async (req, res) => { | ||
| const { incidentId } = req.params; | ||
|
|
||
| res.writeHead(200, { | ||
| 'Content-Type': 'text/event-stream', | ||
| 'Cache-Control': 'no-cache', | ||
| Connection: 'keep-alive', | ||
| 'X-Accel-Buffering': 'no', | ||
| }); | ||
|
|
||
| const incident = await db.incident.findUnique({ | ||
| where: { id: incidentId }, | ||
| }); | ||
|
|
||
| // Emit initial state: stages that already completed before SSE connect | ||
| const initialStages: { stage: PipelineStage; status: 'completed' | 'running'; detail?: string }[] = [ | ||
| { stage: 'TRACE_RECEIVED', status: 'completed', detail: `Trace ${incident?.trace_id ?? ''}` }, | ||
| { stage: 'RISK_ENGINE', status: 'completed', detail: `Severity: ${incident?.severity ?? ''}` }, | ||
| { stage: 'INCIDENT_CREATED', status: 'completed', detail: `Incident ${incidentId}` }, | ||
| ]; | ||
|
|
||
| for (const s of initialStages) { | ||
| res.write(`event: pipeline\ndata: ${JSON.stringify({ incidentId, ...s, timestamp: new Date().toISOString() })}\n\n`); | ||
| } | ||
|
|
||
| // Emit evaluator and subsequent stages based on current status | ||
| const status = incident?.analysis_status ?? 'PENDING'; | ||
| if (status === 'PENDING') { | ||
| res.write(`event: pipeline\ndata: ${JSON.stringify({ incidentId, stage: 'EVALUATOR', status: 'pending', timestamp: new Date().toISOString() })}\n\n`); | ||
| } else if (status === 'PROCESSING') { | ||
| res.write(`event: pipeline\ndata: ${JSON.stringify({ incidentId, stage: 'EVALUATOR', status: 'running', timestamp: new Date().toISOString() })}\n\n`); | ||
| } else if (status === 'COMPLETED' || status === 'FAILED') { | ||
| const s = status === 'COMPLETED' ? 'completed' : 'failed'; | ||
| res.write(`event: pipeline\ndata: ${JSON.stringify({ incidentId, stage: 'EVALUATOR', status: s, timestamp: new Date().toISOString() })}\n\n`); | ||
| res.write(`event: pipeline\ndata: ${JSON.stringify({ incidentId, stage: 'COMPLETED', status: s, timestamp: new Date().toISOString() })}\n\n`); | ||
| } | ||
|
|
||
| const subscriber = createPipelineSubscriber(REDIS_URL); | ||
| subscriber.subscribe(incidentId, (event: PipelineEvent) => { | ||
| res.write(`event: pipeline\ndata: ${JSON.stringify(event)}\n\n`); | ||
| }); | ||
|
|
||
| const keepalive = setInterval(() => { | ||
| res.write(':keepalive\n\n'); | ||
| }, 15000); | ||
|
|
||
| req.on('close', () => { | ||
| subscriber.unsubscribe(incidentId); | ||
| subscriber.quit(); | ||
| clearInterval(keepalive); | ||
| }); | ||
| }); |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Missed-event window between initial-state emission and subscribing; no existence check for incidentId.
The handler reads incident from Postgres, computes/sends the initial synthetic stage events, and only then calls subscriber.subscribe(...). Any pipeline event published by the worker in that gap (e.g., evaluator finishing right as a client connects) is lost forever — Redis pub/sub has no replay, so the client's live view can get stuck showing a stale stage until the next unrelated event happens to fire.
Separately, there's no check that incident is non-null before establishing the stream; a request for a nonexistent incidentId still opens an SSE connection (and a Redis subscription) that will sit idle indefinitely since no worker will ever publish to that channel.
🔧 Suggested fix (subscribe first, validate existence)
+ if (!incident) {
+ res.write(`event: error\ndata: ${JSON.stringify({ message: 'Incident not found' })}\n\n`);
+ return res.end();
+ }
+
+ const subscriber = createPipelineSubscriber(REDIS_URL);
+ subscriber.subscribe(incidentId, (event: PipelineEvent) => {
+ res.write(`event: pipeline\ndata: ${JSON.stringify(event)}\n\n`);
+ });
+
// Emit initial state: stages that already completed before SSE connect
const initialStages: ... = [ ... ];
for (const s of initialStages) { ... }
... // existing evaluator/completed/failed replay logic
-
- const subscriber = createPipelineSubscriber(REDIS_URL);
- subscriber.subscribe(incidentId, (event: PipelineEvent) => {
- res.write(`event: pipeline\ndata: ${JSON.stringify(event)}\n\n`);
- });📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| app.get('/api/investigations/:incidentId/stream', async (req, res) => { | |
| const { incidentId } = req.params; | |
| res.writeHead(200, { | |
| 'Content-Type': 'text/event-stream', | |
| 'Cache-Control': 'no-cache', | |
| Connection: 'keep-alive', | |
| 'X-Accel-Buffering': 'no', | |
| }); | |
| const incident = await db.incident.findUnique({ | |
| where: { id: incidentId }, | |
| }); | |
| // Emit initial state: stages that already completed before SSE connect | |
| const initialStages: { stage: PipelineStage; status: 'completed' | 'running'; detail?: string }[] = [ | |
| { stage: 'TRACE_RECEIVED', status: 'completed', detail: `Trace ${incident?.trace_id ?? ''}` }, | |
| { stage: 'RISK_ENGINE', status: 'completed', detail: `Severity: ${incident?.severity ?? ''}` }, | |
| { stage: 'INCIDENT_CREATED', status: 'completed', detail: `Incident ${incidentId}` }, | |
| ]; | |
| for (const s of initialStages) { | |
| res.write(`event: pipeline\ndata: ${JSON.stringify({ incidentId, ...s, timestamp: new Date().toISOString() })}\n\n`); | |
| } | |
| // Emit evaluator and subsequent stages based on current status | |
| const status = incident?.analysis_status ?? 'PENDING'; | |
| if (status === 'PENDING') { | |
| res.write(`event: pipeline\ndata: ${JSON.stringify({ incidentId, stage: 'EVALUATOR', status: 'pending', timestamp: new Date().toISOString() })}\n\n`); | |
| } else if (status === 'PROCESSING') { | |
| res.write(`event: pipeline\ndata: ${JSON.stringify({ incidentId, stage: 'EVALUATOR', status: 'running', timestamp: new Date().toISOString() })}\n\n`); | |
| } else if (status === 'COMPLETED' || status === 'FAILED') { | |
| const s = status === 'COMPLETED' ? 'completed' : 'failed'; | |
| res.write(`event: pipeline\ndata: ${JSON.stringify({ incidentId, stage: 'EVALUATOR', status: s, timestamp: new Date().toISOString() })}\n\n`); | |
| res.write(`event: pipeline\ndata: ${JSON.stringify({ incidentId, stage: 'COMPLETED', status: s, timestamp: new Date().toISOString() })}\n\n`); | |
| } | |
| const subscriber = createPipelineSubscriber(REDIS_URL); | |
| subscriber.subscribe(incidentId, (event: PipelineEvent) => { | |
| res.write(`event: pipeline\ndata: ${JSON.stringify(event)}\n\n`); | |
| }); | |
| const keepalive = setInterval(() => { | |
| res.write(':keepalive\n\n'); | |
| }, 15000); | |
| req.on('close', () => { | |
| subscriber.unsubscribe(incidentId); | |
| subscriber.quit(); | |
| clearInterval(keepalive); | |
| }); | |
| }); | |
| app.get('/api/investigations/:incidentId/stream', async (req, res) => { | |
| const { incidentId } = req.params; | |
| res.writeHead(200, { | |
| 'Content-Type': 'text/event-stream', | |
| 'Cache-Control': 'no-cache', | |
| Connection: 'keep-alive', | |
| 'X-Accel-Buffering': 'no', | |
| }); | |
| const incident = await db.incident.findUnique({ | |
| where: { id: incidentId }, | |
| }); | |
| if (!incident) { | |
| res.write(`event: error\ndata: ${JSON.stringify({ message: 'Incident not found' })}\n\n`); | |
| return res.end(); | |
| } | |
| const subscriber = createPipelineSubscriber(REDIS_URL); | |
| subscriber.subscribe(incidentId, (event: PipelineEvent) => { | |
| res.write(`event: pipeline\ndata: ${JSON.stringify(event)}\n\n`); | |
| }); | |
| // Emit initial state: stages that already completed before SSE connect | |
| const initialStages: { stage: PipelineStage; status: 'completed' | 'running'; detail?: string }[] = [ | |
| { stage: 'TRACE_RECEIVED', status: 'completed', detail: `Trace ${incident?.trace_id ?? ''}` }, | |
| { stage: 'RISK_ENGINE', status: 'completed', detail: `Severity: ${incident?.severity ?? ''}` }, | |
| { stage: 'INCIDENT_CREATED', status: 'completed', detail: `Incident ${incidentId}` }, | |
| ]; | |
| for (const s of initialStages) { | |
| res.write(`event: pipeline\ndata: ${JSON.stringify({ incidentId, ...s, timestamp: new Date().toISOString() })}\n\n`); | |
| } | |
| // Emit evaluator and subsequent stages based on current status | |
| const status = incident?.analysis_status ?? 'PENDING'; | |
| if (status === 'PENDING') { | |
| res.write(`event: pipeline\ndata: ${JSON.stringify({ incidentId, stage: 'EVALUATOR', status: 'pending', timestamp: new Date().toISOString() })}\n\n`); | |
| } else if (status === 'PROCESSING') { | |
| res.write(`event: pipeline\ndata: ${JSON.stringify({ incidentId, stage: 'EVALUATOR', status: 'running', timestamp: new Date().toISOString() })}\n\n`); | |
| } else if (status === 'COMPLETED' || status === 'FAILED') { | |
| const s = status === 'COMPLETED' ? 'completed' : 'failed'; | |
| res.write(`event: pipeline\ndata: ${JSON.stringify({ incidentId, stage: 'EVALUATOR', status: s, timestamp: new Date().toISOString() })}\n\n`); | |
| res.write(`event: pipeline\ndata: ${JSON.stringify({ incidentId, stage: 'COMPLETED', status: s, timestamp: new Date().toISOString() })}\n\n`); | |
| } | |
| const keepalive = setInterval(() => { | |
| res.write(':keepalive\n\n'); | |
| }, 15000); | |
| req.on('close', () => { | |
| subscriber.unsubscribe(incidentId); | |
| subscriber.quit(); | |
| clearInterval(keepalive); | |
| }); | |
| }); |
🤖 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 `@apps/node-api/src/index.ts` around lines 109 - 160, Update the
`/api/investigations/:incidentId/stream` handler to validate `incident`
immediately after the database lookup and reject nonexistent incidents before
opening the SSE stream or Redis subscription. Establish the
`createPipelineSubscriber` subscription before emitting any initial synthetic
stages, while preserving the existing initial-state payloads and cleanup
behavior so events published during initialization are not missed.
Comment on lines
+223
to
+237
| // Flush only Redis queues (keeps DB intact for fresh demo runs) | ||
| app.post('/api/admin/reset/queue', async (_req, res) => { | ||
| try { | ||
| console.log('[admin] 🧹 Flushing Redis queues...'); | ||
| const redis = new IORedis(REDIS_URL); | ||
| await redis.flushall(); | ||
| redis.disconnect(); | ||
| console.log('[admin] ✅ Redis queues flushed!'); | ||
| res.json({ success: true, message: 'Queues flushed successfully.' }); | ||
| } catch (error) { | ||
| console.error('[admin] ❌ Queue flush error:', error); | ||
| res.status(500).json({ error: 'Failed to flush queues', details: String(error) }); | ||
| } | ||
| }); | ||
|
|
There was a problem hiding this comment.
🔒 Security & Privacy | 🔴 Critical | ⚡ Quick win
Unauthenticated endpoint that flushes the entire Redis database.
POST /api/admin/reset/queue has no authentication/authorization check and calls redis.flushall(), which clears every key in the connected Redis logical database — not just BullMQ queues. Anyone who can reach this route can wipe all queue state, adaptive-sampling windows, and any other Redis-backed data (pub/sub itself is unaffected, but persisted keys are not). The comment "Flushes only Redis queues" is inaccurate for flushall.
🔒 Suggested fix
-app.post('/api/admin/reset/queue', async (_req, res) => {
+app.post('/api/admin/reset/queue', requireAdminAuth, async (_req, res) => {
try {
console.log('[admin] 🧹 Flushing Redis queues...');
const redis = new IORedis(REDIS_URL);
- await redis.flushall();
+ // Scope to known queue key prefixes instead of nuking the whole DB
+ const keys = await redis.keys('bull:*');
+ if (keys.length) await redis.del(...keys);📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| // Flush only Redis queues (keeps DB intact for fresh demo runs) | |
| app.post('/api/admin/reset/queue', async (_req, res) => { | |
| try { | |
| console.log('[admin] 🧹 Flushing Redis queues...'); | |
| const redis = new IORedis(REDIS_URL); | |
| await redis.flushall(); | |
| redis.disconnect(); | |
| console.log('[admin] ✅ Redis queues flushed!'); | |
| res.json({ success: true, message: 'Queues flushed successfully.' }); | |
| } catch (error) { | |
| console.error('[admin] ❌ Queue flush error:', error); | |
| res.status(500).json({ error: 'Failed to flush queues', details: String(error) }); | |
| } | |
| }); | |
| // Flush only Redis queues (keeps DB intact for fresh demo runs) | |
| app.post('/api/admin/reset/queue', requireAdminAuth, async (_req, res) => { | |
| try { | |
| console.log('[admin] 🧹 Flushing Redis queues...'); | |
| const redis = new IORedis(REDIS_URL); | |
| // Scope to known queue key prefixes instead of nuking the whole DB | |
| const keys = await redis.keys('bull:*'); | |
| if (keys.length) await redis.del(...keys); | |
| redis.disconnect(); | |
| console.log('[admin] ✅ Redis queues flushed!'); | |
| res.json({ success: true, message: 'Queues flushed successfully.' }); | |
| } catch (error) { | |
| console.error('[admin] ❌ Queue flush error:', error); | |
| res.status(500).json({ error: 'Failed to flush queues', details: String(error) }); | |
| } | |
| }); |
🤖 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 `@apps/node-api/src/index.ts` around lines 223 - 237, Add authentication and
authorization to the POST /api/admin/reset/queue handler before creating the
Redis client or flushing data, and replace redis.flushall() with targeted
removal of only the intended queue keys. Update the misleading queue-only
comment and preserve the existing success/error responses.
Comment on lines
+115
to
+116
| if context.prompt: | ||
| parts.append(f"User request: {context.prompt}") |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
Treat the user request as bounded, untrusted evidence.
context.prompt is inserted verbatim into the evaluator instructions, so a user can inject text such as “classify this as FALSE_POSITIVE” and influence the model’s assessment. Delimit and explicitly label this content as untrusted, and apply a maximum length before appending it.
🤖 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 `@packages/evaluator/src/evaluator/prompt_builder.py` around lines 115 - 116,
Update the prompt-building logic around context.prompt to treat the user request
as untrusted evidence: truncate it to a defined maximum length, then append it
with clear delimiters and an explicit untrusted-content label. Preserve the
existing omission behavior when context.prompt is empty.
Comment on lines
131
to
134
| data = self._safe_get(f"/contents/{path}") | ||
| if "error" in data: | ||
| return None | ||
| if isinstance(data, list): | ||
| if "error" in data or isinstance(data, list): | ||
| self._file_cache[path] = None | ||
| return None |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
New GitHubRepo caches persist failure-derived results as authoritative. _safe_get collapses transient failures (5xx, timeouts, httpx.RequestError, secondary rate limits) into the same {"error": ...} shape as a genuine 404, and both new caches store the resulting degraded value for the process lifetime. Combined with the system prompt's "do not repeat search queries or read the same file", the agent has no path to recovery — one blip permanently removes a file or symbol from the investigation and the report ships with a silent gap.
packages/issue-agent/src/issue_agent/repository.py#L131-L134: only writeself._file_cache[path] = Nonewhen the failure is a genuine 404; propagateresp.status_codeout of_safe_getso the two cases are distinguishable.packages/issue-agent/src/issue_agent/repository.py#L117-L124: skipself._search_cache[symbol] = resultwhen bothcontent_dataandtree_datacarry"error", so an empty match list caused by API failure isn't cached as a real negative.
📍 Affects 1 file
packages/issue-agent/src/issue_agent/repository.py#L131-L134(this comment)packages/issue-agent/src/issue_agent/repository.py#L117-L124
🤖 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 `@packages/issue-agent/src/issue_agent/repository.py` around lines 131 - 134,
Update GitHubRepo._safe_get to preserve the HTTP status code for failures, then
in the file-loading cache path at
packages/issue-agent/src/issue_agent/repository.py:131-134 cache None only for
genuine 404 responses; transient failures must not be cached. In the search
caching path at packages/issue-agent/src/issue_agent/repository.py:117-124, skip
self._search_cache[symbol] = result when both content_data and tree_data contain
"error", allowing later attempts to recover.
Comment on lines
229
to
235
| try: | ||
| return full.read_text(encoding="utf-8") | ||
| content = full.read_text(encoding="utf-8") | ||
| self._file_cache[path] = content | ||
| return content | ||
| except (UnicodeDecodeError, ValueError): | ||
| self._file_cache[path] = None | ||
| return None |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Catch OSError too.
read_text can raise PermissionError (an OSError), which isn't covered by (UnicodeDecodeError, ValueError) and would escape the tool call. Note UnicodeDecodeError is a ValueError subclass, so the tuple can collapse.
🐛 Proposed fix
try:
content = full.read_text(encoding="utf-8")
self._file_cache[path] = content
return content
- except (UnicodeDecodeError, ValueError):
+ except (OSError, ValueError):
self._file_cache[path] = None
return None📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| try: | |
| return full.read_text(encoding="utf-8") | |
| content = full.read_text(encoding="utf-8") | |
| self._file_cache[path] = content | |
| return content | |
| except (UnicodeDecodeError, ValueError): | |
| self._file_cache[path] = None | |
| return None | |
| try: | |
| content = full.read_text(encoding="utf-8") | |
| self._file_cache[path] = content | |
| return content | |
| except (OSError, ValueError): | |
| self._file_cache[path] = None | |
| return None |
🤖 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 `@packages/issue-agent/src/issue_agent/repository.py` around lines 229 - 235,
Update the exception handling around the read_text call in the repository
file-reading method to catch OSError alongside ValueError, allowing filesystem
failures such as PermissionError to follow the existing None-cache and return
path. Simplify the exception tuple since UnicodeDecodeError is already covered
by ValueError.
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.
… live
Summary by cubic
Add SSE-powered live updates for investigations so the frontend can see pipeline progress in real time. Also tightened evaluator and issue-agent behavior to reduce cost and improve reliability.
New Features
/api/investigations/:incidentId/stream(SSE) streamingpipelineevents with stage, status, detail; includes initial catch-up and 15s keepalive.pipelinemodule;apps/node-apiand worker now emit stages:RISK_ENGINE,INCIDENT_CREATED,EVALUATOR,PROMOTION_GATE,ISSUE_AGENT(with sub-steps),COMPLETED.traceId,signozTraceUrl, anderrorCodeon failures; added/api/admin/reset/queueto flush Redis queues only.Refactors
packages/incident-formation: only re-queue evaluation on escalate-to-CRITICAL, failed, or pending; always refresh occurrence/metadata.packages/adaptive-sampling: returns sampled execution id;/api/tracesechoes that id and asampledboolean; sampling consumer creates a fallback incident when evaluator fails.packages/evaluator: context now includesprompt; prompt shows the user request; refinedTOOL_CALL_ANOMALYguidance.packages/issue-agent: enforce 8-call tool budget, cap file reads to 14KB, add repo/search/graph caches, reduce retries to 2, and always print JSON (addsissue_urlwhen created).Written for commit 6d8e741. Summary will update on new commits.
Summary by CodeRabbit
New Features
Bug Fixes