mirror of
https://github.com/Sea-Haven-Industries/open-swe.git
synced 2026-09-30 17:23:15 +00:00
2b01652754
* Adopt upstream modular webhook skeleton (#1621) Apply the durable-interrupt-dispatch refactor: split the monolithic webapp.py into a thin routing layer plus per-source handlers in webhooks/{github,slack,linear}.py, and add completion.py, dispatch.py, and reconcile.py. Reconcile fork divergence by keeping the Bedrock/ Fireworks cross-provider fallback, the no-agent-attribution prompt policy, the dashboard-handoff re-export, and the Slack channel-info cache. ci_autofix is restored on the new dispatch model in a later commit. Refs: #80 * Port fork webhook security delta onto modular handlers Re-apply the fork's security customizations that #1621 did not carry: Linear webhook replay protection (freshness window on the signed webhookTimestamp), per-repo token-cache binding threaded through the thread token resolvers, the INTERNAL_BOT_LOGINS self-check in the review-finding-reply path, and a user-mapping cache refresh before email resolution on the issue and PR-comment paths (multi-replica staleness). Existing fork security tests pass unchanged. Refs: #80 * Restore CI auto-fix on the modular dispatch model Bring back ci_autofix.py and the ci_monitor graph that #1621 deleted, re-wiring the fork's security-reviewed PR-babysitting onto the new structure: the CI-event, autofix-toggle, and review-feedback handlers move into webhooks/github.py and the github_webhook router re-gains the check_run/check_suite/workflow_run/status routing plus the autofix command and actionable-review branches. Auto-fix runs now dispatch through dispatch_agent_run (durability + completion webhook) while keeping the deliberate batch-while-busy skip-rule via get_thread_active_status. Restore langgraph.json's ci_monitor entry and the fork autofix tests (dispatch mock + import paths re-pointed). Refs: #80 * Reformat and update docs for the modular webhook split Point CLAUDE.md and deploy/MIGRATION.md at the new webhooks/ modules and the dispatch/completion/reconcile contract, and mark the user-mapping cache-refresh fix as applied on the GitHub handlers. Refs: #80 * Restore reject backstop for autofix dispatch A burst of near-simultaneous CI events for one head SHA can slip past the busy-check before the dedupe SHA is recorded, so dispatch the autofix path with multitask_strategy=reject (dev's prior platform default) to drop duplicate concurrent creates instead of letting them interrupt each other. Also make the completion failure-reply dedup claim-then-post and drop the unreachable interrupted branch. --------- Co-authored-by: amoussa1229 <166072409+amoussa1229@users.noreply.github.com>
84 lines
3 KiB
Python
84 lines
3 KiB
Python
"""Shared LangGraph thread helpers for the dashboard.
|
|
|
|
The webhook triggers (Slack / Linear / GitHub) dispatch through
|
|
``agent.dispatch.dispatch_agent_run`` with ``multitask_strategy="interrupt"``,
|
|
so they no longer need a busy-check or an in-process lock. The store-queue
|
|
below is retained for the dashboard's deliberate "inject a follow-up into a
|
|
run that's already in flight" path (``thread_api.send_dashboard_message``).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
from typing import Any
|
|
|
|
from langgraph_sdk import get_client
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
MAX_QUEUED_MESSAGES = 100
|
|
|
|
|
|
def langgraph_url() -> str:
|
|
return os.environ.get("LANGGRAPH_URL") or os.environ.get(
|
|
"LANGGRAPH_URL_PROD", "http://localhost:2024"
|
|
)
|
|
|
|
|
|
def langgraph_client():
|
|
return get_client(url=langgraph_url())
|
|
|
|
|
|
async def get_thread_active_status(thread_id: str) -> bool | None:
|
|
"""Return whether the thread is active, or None when status cannot be determined."""
|
|
try:
|
|
thread = await langgraph_client().threads.get(thread_id)
|
|
status = thread.get("status", "idle") if isinstance(thread, dict) else "idle"
|
|
logger.info("Thread %s status check: status=%s", thread_id, status)
|
|
return status == "busy"
|
|
except Exception as exc: # noqa: BLE001
|
|
logger.warning("Failed to get thread status for %s: %s", thread_id, exc)
|
|
return None
|
|
|
|
|
|
async def queue_message_for_thread(
|
|
thread_id: str, message_content: str | list[dict[str, Any]] | dict[str, Any]
|
|
) -> bool:
|
|
"""Queue a follow-up message for a busy thread (FIFO store namespace).
|
|
|
|
Used by the dashboard to inject a follow-up into a run that's already in
|
|
flight; webhook triggers use ``multitask_strategy="interrupt"`` instead.
|
|
"""
|
|
client = langgraph_client()
|
|
try:
|
|
namespace = ("queue", thread_id)
|
|
key = "pending_messages"
|
|
new_message = {"content": message_content}
|
|
|
|
existing_messages: list[dict[str, Any]] = []
|
|
try:
|
|
existing_item = await client.store.get_item(namespace, key)
|
|
if existing_item and existing_item.get("value"):
|
|
existing_messages = existing_item["value"].get("messages", [])
|
|
except Exception: # noqa: BLE001
|
|
logger.debug("No existing queued messages for thread %s", thread_id)
|
|
|
|
existing_messages.append(new_message)
|
|
if len(existing_messages) > MAX_QUEUED_MESSAGES:
|
|
existing_messages = existing_messages[-MAX_QUEUED_MESSAGES:]
|
|
logger.warning(
|
|
"Thread %s queue capped at %d messages (dropped oldest)",
|
|
thread_id,
|
|
MAX_QUEUED_MESSAGES,
|
|
)
|
|
await client.store.put_item(namespace, key, {"messages": existing_messages})
|
|
logger.info(
|
|
"Queued message for thread %s (total queued: %d)",
|
|
thread_id,
|
|
len(existing_messages),
|
|
)
|
|
return True
|
|
except Exception:
|
|
logger.exception("Failed to queue message for thread %s", thread_id)
|
|
return False
|