724 lines
26 KiB
Python
724 lines
26 KiB
Python
"""Gateway API for the AgentScope report-collaboration workbench.
|
|
|
|
Plan/run writes enqueue durable jobs and return immediately. AgentScope
|
|
execution is not started from this router. ``GET /runs/{id}/stream`` is a
|
|
read-only durable SSE tail: disconnect does not cancel the run.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, Header, HTTPException, Query, Request
|
|
from fastapi.responses import JSONResponse, Response, StreamingResponse
|
|
from pydantic import BaseModel, Field
|
|
|
|
from app.gateway.deps import get_current_user
|
|
from app.report_collaboration.agentscope_runtime.model_resolver import freeze_plan_models
|
|
from app.report_collaboration.contracts.snapshots import SessionSnapshot
|
|
from app.report_collaboration.execution.live_hub import iter_run_sse, replay_cursor
|
|
from app.report_collaboration.interventions.intent_router import build_runtime_intent_router
|
|
from app.report_collaboration.interventions.service import RuntimeCommandService
|
|
from app.report_collaboration.planning import PlanProposalService
|
|
from app.report_collaboration.reporting import ConservativeRewriteGenerator, ModelRewriteGenerator, ReportRewriteService, freeze_report_template
|
|
from app.report_collaboration.reporting.templates import requirement_snapshot_from_record
|
|
from app.report_collaboration.requirement import RequirementResolver
|
|
from app.report_collaboration.security import freeze_run_budget
|
|
from deerflow.config.app_config import get_app_config
|
|
from deerflow.persistence.report_collaboration import (
|
|
ReportCollaborationConflictError,
|
|
ReportCollaborationLimitError,
|
|
ReportCollaborationNotFoundError,
|
|
ReportCollaborationStore,
|
|
ReportCollaborationValidationError,
|
|
)
|
|
|
|
router = APIRouter(prefix="/api/report-collaboration", tags=["report-collaboration"])
|
|
|
|
|
|
class CreateSessionBody(BaseModel):
|
|
title: str | None = Field(default=None, max_length=255)
|
|
|
|
|
|
class UpdateSessionBody(BaseModel):
|
|
title: str | None = Field(default=None, max_length=255)
|
|
|
|
|
|
class SendMessageBody(BaseModel):
|
|
text: str = Field(..., min_length=1, max_length=20_000)
|
|
target_node_id: str | None = None
|
|
agent_run_id: str | None = None
|
|
|
|
|
|
class StartRunBody(BaseModel):
|
|
plan_id: str = Field(..., min_length=1, max_length=64)
|
|
|
|
|
|
class CreateCommandBody(BaseModel):
|
|
text: str = Field(..., min_length=1, max_length=20_000)
|
|
target_node_id: str | None = None
|
|
agent_run_id: str | None = None
|
|
|
|
|
|
class CreateRewriteBody(BaseModel):
|
|
instruction: str = Field(..., min_length=1, max_length=20_000)
|
|
section: str | None = None
|
|
|
|
|
|
def _store(request: Request) -> ReportCollaborationStore:
|
|
store = getattr(request.app.state, "report_collaboration_store", None)
|
|
if store is None:
|
|
raise HTTPException(status_code=503, detail="Report collaboration store is unavailable")
|
|
return store
|
|
|
|
|
|
def _runtime_commands(request: Request) -> RuntimeCommandService:
|
|
service = getattr(request.app.state, "report_collaboration_command_service", None)
|
|
if service is None:
|
|
app_config = get_app_config()
|
|
service = RuntimeCommandService(
|
|
_store(request),
|
|
router=build_runtime_intent_router(app_config),
|
|
config=app_config.report_collaboration,
|
|
)
|
|
request.app.state.report_collaboration_command_service = service
|
|
return service
|
|
|
|
|
|
def _require_enabled() -> JSONResponse | None:
|
|
if not get_app_config().report_collaboration.enabled:
|
|
return _error(503, "REPORT_COLLABORATION_DISABLED", "报告协作工作台未启用")
|
|
return None
|
|
|
|
|
|
async def _require_user(request: Request) -> str:
|
|
user_id = await get_current_user(request)
|
|
if not user_id:
|
|
raise HTTPException(status_code=401, detail="Authentication required")
|
|
return str(user_id)
|
|
|
|
|
|
def _error(status: int, code: str, message: str, detail: Any = None) -> JSONResponse:
|
|
body: dict[str, Any] = {"code": code, "message": message}
|
|
if detail is not None:
|
|
body["detail"] = detail
|
|
return JSONResponse(status_code=status, content=body)
|
|
|
|
|
|
def _require_idempotency(idempotency_key: str | None) -> str | JSONResponse:
|
|
key = (idempotency_key or "").strip()
|
|
if not key:
|
|
return _error(422, "IDEMPOTENCY_KEY_REQUIRED", "写操作必须携带 X-Idempotency-Key")
|
|
if len(key) > 128:
|
|
return _error(422, "IDEMPOTENCY_KEY_INVALID", "X-Idempotency-Key 过长")
|
|
return key
|
|
|
|
|
|
def _map_store_error(exc: Exception) -> JSONResponse:
|
|
if isinstance(exc, ReportCollaborationNotFoundError):
|
|
return _error(404, "NOT_FOUND", str(exc))
|
|
if isinstance(exc, ReportCollaborationConflictError):
|
|
return _error(409, exc.code, str(exc), detail={"current_revision": exc.current_revision})
|
|
if isinstance(exc, ReportCollaborationValidationError):
|
|
return _error(422, exc.code, str(exc))
|
|
if isinstance(exc, ReportCollaborationLimitError):
|
|
return _error(429, exc.code, str(exc), detail={"usage": exc.usage, "limits": exc.limits})
|
|
raise exc
|
|
|
|
|
|
def _rewrite_service(store: ReportCollaborationStore) -> ReportRewriteService:
|
|
cfg = get_app_config().report_collaboration
|
|
generator = ModelRewriteGenerator(cfg.writer_model) if cfg.writer_model else ConservativeRewriteGenerator()
|
|
return ReportRewriteService(store, generator=generator)
|
|
|
|
|
|
async def _frozen_run_template(request: Request, session: dict[str, Any], plan: dict[str, Any]) -> dict[str, Any]:
|
|
requirement = requirement_snapshot_from_record(session)
|
|
structure_id = plan.get("report_structure_id") or (requirement.report_structure_id if requirement else None)
|
|
structure = None
|
|
structure_store = getattr(request.app.state, "report_structure_store", None)
|
|
if structure_id and structure_store is not None:
|
|
structure = await structure_store.get_structure(str(structure_id))
|
|
return freeze_report_template(structure, requirement=requirement).model_dump(mode="json")
|
|
|
|
|
|
async def _owned_session(store: ReportCollaborationStore, session_id: str, user_id: str) -> dict[str, Any] | JSONResponse:
|
|
record = await store.get_session_record(session_id)
|
|
if record is None or record.get("owner_id") != user_id:
|
|
return _error(404, "NOT_FOUND", f"report collaboration session not found: {session_id}")
|
|
return record
|
|
|
|
|
|
async def _owned_run(store: ReportCollaborationStore, run_id: str, user_id: str) -> dict[str, Any] | JSONResponse:
|
|
run = await store.get_run(run_id)
|
|
if run is None:
|
|
return _error(404, "NOT_FOUND", f"report collaboration run not found: {run_id}")
|
|
owned = await _owned_session(store, run["session_id"], user_id)
|
|
if isinstance(owned, JSONResponse):
|
|
return owned
|
|
return run
|
|
|
|
|
|
@router.post("/sessions")
|
|
async def create_session(
|
|
request: Request,
|
|
body: CreateSessionBody | None = None,
|
|
x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
|
|
):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
key = _require_idempotency(x_idempotency_key)
|
|
if isinstance(key, JSONResponse):
|
|
return key
|
|
payload = body or CreateSessionBody()
|
|
try:
|
|
return await _store(request).create_session(owner_id=user_id, title=payload.title or "", idempotency_key=key)
|
|
except Exception as exc:
|
|
return _map_store_error(exc)
|
|
|
|
|
|
@router.get("/sessions")
|
|
async def list_sessions(request: Request):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
sessions = await _store(request).list_sessions(user_id)
|
|
return {"sessions": sessions}
|
|
|
|
|
|
@router.get("/sessions/{session_id}")
|
|
async def get_session_snapshot(request: Request, session_id: str):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
store = _store(request)
|
|
owned = await _owned_session(store, session_id, user_id)
|
|
if isinstance(owned, JSONResponse):
|
|
return owned
|
|
snapshot = await store.get_snapshot(session_id)
|
|
return SessionSnapshot.model_validate(snapshot).model_dump()
|
|
|
|
|
|
@router.patch("/sessions/{session_id}")
|
|
async def update_session(
|
|
request: Request,
|
|
session_id: str,
|
|
body: UpdateSessionBody,
|
|
x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
|
|
x_expected_revision: int | None = Header(default=None, alias="X-Expected-Revision"),
|
|
):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
key = _require_idempotency(x_idempotency_key)
|
|
if isinstance(key, JSONResponse):
|
|
return key
|
|
store = _store(request)
|
|
owned = await _owned_session(store, session_id, user_id)
|
|
if isinstance(owned, JSONResponse):
|
|
return owned
|
|
try:
|
|
updated = await store.update_session(session_id, title=body.title, expected_revision=x_expected_revision)
|
|
await store.create_command(session_id, idempotency_key=key, text=body.title or "", operation="update_session", status="completed")
|
|
return updated
|
|
except Exception as exc:
|
|
return _map_store_error(exc)
|
|
|
|
|
|
@router.delete("/sessions/{session_id}")
|
|
async def delete_session(request: Request, session_id: str):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
store = _store(request)
|
|
owned = await _owned_session(store, session_id, user_id)
|
|
if isinstance(owned, JSONResponse):
|
|
return owned
|
|
await store.delete_session(session_id)
|
|
return Response(status_code=204)
|
|
|
|
|
|
@router.get("/sessions/{session_id}/messages")
|
|
async def list_messages(request: Request, session_id: str):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
store = _store(request)
|
|
owned = await _owned_session(store, session_id, user_id)
|
|
if isinstance(owned, JSONResponse):
|
|
return owned
|
|
return {"messages": await store.list_messages(session_id)}
|
|
|
|
|
|
@router.post("/sessions/{session_id}/messages")
|
|
async def send_message(
|
|
request: Request,
|
|
session_id: str,
|
|
body: SendMessageBody,
|
|
x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
|
|
):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
key = _require_idempotency(x_idempotency_key)
|
|
if isinstance(key, JSONResponse):
|
|
return key
|
|
store = _store(request)
|
|
owned = await _owned_session(store, session_id, user_id)
|
|
if isinstance(owned, JSONResponse):
|
|
return owned
|
|
metadata: dict[str, Any] = {"idempotency_key": key}
|
|
if body.target_node_id:
|
|
metadata["target_node_id"] = body.target_node_id
|
|
if body.agent_run_id:
|
|
metadata["agent_run_id"] = body.agent_run_id
|
|
try:
|
|
existing = await store.get_message_by_idempotency(key)
|
|
if existing is not None:
|
|
return existing
|
|
snapshot = await store.get_snapshot(session_id)
|
|
active = snapshot.get("run") or {}
|
|
if str(active.get("status") or "") in {
|
|
"queued",
|
|
"planning",
|
|
"running",
|
|
"awaiting_input",
|
|
"reviewing",
|
|
"recovering",
|
|
}:
|
|
return _error(
|
|
409,
|
|
"RUN_ACTIVE",
|
|
"运行进行中请使用 /runs/{run_id}/commands 发送干预，不要走规划前消息接口",
|
|
detail={"run_id": active.get("id"), "status": active.get("status")},
|
|
)
|
|
created = await store.create_message(
|
|
session_id,
|
|
role="human",
|
|
content=body.text,
|
|
idempotency_key=key,
|
|
metadata=metadata,
|
|
agent_run_id=body.agent_run_id,
|
|
node_run_id=body.target_node_id,
|
|
)
|
|
await RequirementResolver(store).handle_user_message(session_id, user_message=created, idempotency_key=key)
|
|
return created
|
|
except Exception as exc:
|
|
return _map_store_error(exc)
|
|
|
|
|
|
@router.post("/sessions/{session_id}/plan-requests")
|
|
async def request_plans(
|
|
request: Request,
|
|
session_id: str,
|
|
x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
|
|
x_expected_revision: int | None = Header(default=None, alias="X-Expected-Revision"),
|
|
):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
key = _require_idempotency(x_idempotency_key)
|
|
if isinstance(key, JSONResponse):
|
|
return key
|
|
store = _store(request)
|
|
owned = await _owned_session(store, session_id, user_id)
|
|
if isinstance(owned, JSONResponse):
|
|
return owned
|
|
if owned.get("selected_plan_id"):
|
|
return _error(409, "PLAN_ALREADY_SELECTED", "会话已选定方案，无法重新规划")
|
|
try:
|
|
command = await store.enqueue_plan_request(session_id, idempotency_key=key, expected_revision=x_expected_revision)
|
|
if command.get("status") == "pending":
|
|
await PlanProposalService(
|
|
store,
|
|
agent_store=getattr(request.app.state, "agent_store", None),
|
|
).fulfill(session_id, user_id=user_id, command_id=command["command_id"], idempotency_key=key)
|
|
return Response(status_code=204)
|
|
except Exception as exc:
|
|
return _map_store_error(exc)
|
|
|
|
|
|
@router.get("/sessions/{session_id}/plans")
|
|
async def list_plans(request: Request, session_id: str):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
store = _store(request)
|
|
owned = await _owned_session(store, session_id, user_id)
|
|
if isinstance(owned, JSONResponse):
|
|
return owned
|
|
return {"plans": await store.list_plans(session_id)}
|
|
|
|
|
|
@router.post("/sessions/{session_id}/plans/{plan_id}/select")
|
|
async def select_plan(
|
|
request: Request,
|
|
session_id: str,
|
|
plan_id: str,
|
|
x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
|
|
x_expected_revision: int | None = Header(default=None, alias="X-Expected-Revision"),
|
|
):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
key = _require_idempotency(x_idempotency_key)
|
|
if isinstance(key, JSONResponse):
|
|
return key
|
|
store = _store(request)
|
|
owned = await _owned_session(store, session_id, user_id)
|
|
if isinstance(owned, JSONResponse):
|
|
return owned
|
|
try:
|
|
result = await store.select_plan(session_id, plan_id, idempotency_key=key, expected_revision=x_expected_revision)
|
|
return {"selected_plan_id": result["selected_plan_id"]}
|
|
except Exception as exc:
|
|
return _map_store_error(exc)
|
|
|
|
|
|
@router.post("/sessions/{session_id}/runs")
|
|
async def start_run(
|
|
request: Request,
|
|
session_id: str,
|
|
body: StartRunBody,
|
|
x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
|
|
x_expected_revision: int | None = Header(default=None, alias="X-Expected-Revision"),
|
|
):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
key = _require_idempotency(x_idempotency_key)
|
|
if isinstance(key, JSONResponse):
|
|
return key
|
|
store = _store(request)
|
|
owned = await _owned_session(store, session_id, user_id)
|
|
if isinstance(owned, JSONResponse):
|
|
return owned
|
|
try:
|
|
plans = await store.list_plans(session_id)
|
|
plan = next((item for item in plans if item.get("id") == body.plan_id), None)
|
|
if plan is None:
|
|
return _error(404, "NOT_FOUND", f"report collaboration plan not found: {body.plan_id}")
|
|
bundle = freeze_plan_models(plan, get_app_config())
|
|
cfg = get_app_config().report_collaboration
|
|
run = await store.create_run(
|
|
session_id,
|
|
plan_id=body.plan_id,
|
|
idempotency_key=key,
|
|
expected_revision=x_expected_revision,
|
|
role_snapshots=bundle.dump_by_role(),
|
|
template_snapshot=await _frozen_run_template(request, owned, plan),
|
|
budget_snapshot=freeze_run_budget(cfg),
|
|
max_active_runs_per_owner=cfg.max_concurrent_runs_per_user,
|
|
)
|
|
existing_events = await store.list_events(run["id"], after_seq=0, limit=1)
|
|
if not existing_events:
|
|
await store.append_event(
|
|
session_id=session_id,
|
|
run_id=run["id"],
|
|
event_type="heartbeat",
|
|
data={
|
|
"audit": "model_snapshot",
|
|
"roles": [item.model_dump() for item in bundle.by_role.values()],
|
|
"fallbacks": [item.model_dump() for item in bundle.fallbacks()],
|
|
},
|
|
)
|
|
return run
|
|
except Exception as exc:
|
|
return _map_store_error(exc)
|
|
|
|
|
|
@router.get("/sessions/{session_id}/reports")
|
|
async def list_reports(request: Request, session_id: str):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
store = _store(request)
|
|
owned = await _owned_session(store, session_id, user_id)
|
|
if isinstance(owned, JSONResponse):
|
|
return owned
|
|
return {"versions": await store.list_report_versions(session_id)}
|
|
|
|
|
|
@router.post("/sessions/{session_id}/report-rewrites")
|
|
async def create_rewrite(
|
|
request: Request,
|
|
session_id: str,
|
|
body: CreateRewriteBody,
|
|
x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
|
|
):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
key = _require_idempotency(x_idempotency_key)
|
|
if isinstance(key, JSONResponse):
|
|
return key
|
|
store = _store(request)
|
|
owned = await _owned_session(store, session_id, user_id)
|
|
if isinstance(owned, JSONResponse):
|
|
return owned
|
|
try:
|
|
return await _rewrite_service(store).create_candidate(
|
|
session_id,
|
|
idempotency_key=key,
|
|
instruction=body.instruction,
|
|
section=body.section,
|
|
)
|
|
except Exception as exc:
|
|
return _map_store_error(exc)
|
|
|
|
|
|
@router.post("/sessions/{session_id}/report-rewrites/{rewrite_id}/apply")
|
|
async def apply_rewrite(
|
|
request: Request,
|
|
session_id: str,
|
|
rewrite_id: str,
|
|
x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
|
|
):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
key = _require_idempotency(x_idempotency_key)
|
|
if isinstance(key, JSONResponse):
|
|
return key
|
|
store = _store(request)
|
|
owned = await _owned_session(store, session_id, user_id)
|
|
if isinstance(owned, JSONResponse):
|
|
return owned
|
|
try:
|
|
version, _created = await _rewrite_service(store).apply_candidate(session_id, rewrite_id, idempotency_key=key)
|
|
return version
|
|
except Exception as exc:
|
|
return _map_store_error(exc)
|
|
|
|
|
|
@router.post("/sessions/{session_id}/report-versions/{version_id}/restore")
|
|
async def restore_report(
|
|
request: Request,
|
|
session_id: str,
|
|
version_id: str,
|
|
x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
|
|
):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
key = _require_idempotency(x_idempotency_key)
|
|
if isinstance(key, JSONResponse):
|
|
return key
|
|
store = _store(request)
|
|
owned = await _owned_session(store, session_id, user_id)
|
|
if isinstance(owned, JSONResponse):
|
|
return owned
|
|
try:
|
|
version, _created = await _rewrite_service(store).restore(session_id, version_id, idempotency_key=key)
|
|
return version
|
|
except Exception as exc:
|
|
return _map_store_error(exc)
|
|
|
|
|
|
@router.get("/runs/{run_id}")
|
|
async def get_run(request: Request, run_id: str):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
run = await _owned_run(_store(request), run_id, user_id)
|
|
if isinstance(run, JSONResponse):
|
|
return run
|
|
return run
|
|
|
|
|
|
@router.get("/runs/{run_id}/events")
|
|
async def list_run_events(
|
|
request: Request,
|
|
run_id: str,
|
|
after_seq: int = Query(default=0, ge=0),
|
|
limit: int = Query(default=200, ge=1, le=1000),
|
|
):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
store = _store(request)
|
|
run = await _owned_run(store, run_id, user_id)
|
|
if isinstance(run, JSONResponse):
|
|
return run
|
|
return {"events": await store.list_events(run_id, after_seq=after_seq, limit=limit)}
|
|
|
|
|
|
@router.get("/runs/{run_id}/stream")
|
|
async def stream_run(
|
|
request: Request,
|
|
run_id: str,
|
|
after_seq: int = Query(default=0, ge=0),
|
|
last_event_id: str | None = Header(default=None, alias="Last-Event-ID"),
|
|
):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
store = _store(request)
|
|
run = await _owned_run(store, run_id, user_id)
|
|
if isinstance(run, JSONResponse):
|
|
return run
|
|
hub = getattr(request.app.state, "report_collaboration_live_hub", None)
|
|
cursor = replay_cursor(last_event_id or request.headers.get("last-event-id"), after_seq)
|
|
settings = getattr(request.app.state, "report_collaboration_sse_settings", None)
|
|
|
|
async def generator():
|
|
async def _disconnected() -> bool:
|
|
return await request.is_disconnected()
|
|
|
|
async for chunk in iter_run_sse(
|
|
store=store,
|
|
hub=hub,
|
|
run_id=run_id,
|
|
cursor=cursor,
|
|
is_disconnected=_disconnected,
|
|
settings=settings,
|
|
):
|
|
yield chunk
|
|
|
|
return StreamingResponse(
|
|
generator(),
|
|
media_type="text/event-stream",
|
|
headers={
|
|
"Cache-Control": "no-cache, no-transform",
|
|
"Connection": "keep-alive",
|
|
"X-Accel-Buffering": "no",
|
|
"X-Report-Collaboration-Run-Status": str(run.get("status") or ""),
|
|
},
|
|
)
|
|
|
|
|
|
@router.post("/runs/{run_id}/cancel")
|
|
async def cancel_run(
|
|
request: Request,
|
|
run_id: str,
|
|
x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
|
|
):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
key = _require_idempotency(x_idempotency_key)
|
|
if isinstance(key, JSONResponse):
|
|
return key
|
|
store = _store(request)
|
|
run = await _owned_run(store, run_id, user_id)
|
|
if isinstance(run, JSONResponse):
|
|
return run
|
|
try:
|
|
return await store.request_cancel_run(run_id, idempotency_key=key)
|
|
except Exception as exc:
|
|
return _map_store_error(exc)
|
|
|
|
|
|
@router.post("/runs/{run_id}/commands")
|
|
async def create_command(
|
|
request: Request,
|
|
run_id: str,
|
|
body: CreateCommandBody,
|
|
x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
|
|
):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
key = _require_idempotency(x_idempotency_key)
|
|
if isinstance(key, JSONResponse):
|
|
return key
|
|
store = _store(request)
|
|
run = await _owned_run(store, run_id, user_id)
|
|
if isinstance(run, JSONResponse):
|
|
return run
|
|
try:
|
|
return await _runtime_commands(request).submit(
|
|
run_id,
|
|
text=body.text,
|
|
idempotency_key=key,
|
|
target_node_id=body.target_node_id,
|
|
agent_run_id=body.agent_run_id,
|
|
)
|
|
except Exception as exc:
|
|
return _map_store_error(exc)
|
|
|
|
|
|
@router.post("/runs/{run_id}/commands/{command_id}/confirm")
|
|
async def confirm_command(
|
|
request: Request,
|
|
run_id: str,
|
|
command_id: str,
|
|
x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
|
|
):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
key = _require_idempotency(x_idempotency_key)
|
|
if isinstance(key, JSONResponse):
|
|
return key
|
|
store = _store(request)
|
|
run = await _owned_run(store, run_id, user_id)
|
|
if isinstance(run, JSONResponse):
|
|
return run
|
|
try:
|
|
return await _runtime_commands(request).confirm(run_id, command_id, idempotency_key=key)
|
|
except Exception as exc:
|
|
return _map_store_error(exc)
|
|
|
|
|
|
@router.post("/runs/{run_id}/commands/{command_id}/cancel")
|
|
async def cancel_command(
|
|
request: Request,
|
|
run_id: str,
|
|
command_id: str,
|
|
x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
|
|
):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
key = _require_idempotency(x_idempotency_key)
|
|
if isinstance(key, JSONResponse):
|
|
return key
|
|
store = _store(request)
|
|
run = await _owned_run(store, run_id, user_id)
|
|
if isinstance(run, JSONResponse):
|
|
return run
|
|
try:
|
|
return await _runtime_commands(request).cancel(run_id, command_id, idempotency_key=key)
|
|
except Exception as exc:
|
|
return _map_store_error(exc)
|
|
|
|
|
|
@router.get("/runs/{run_id}/nodes/{node_id}/contract")
|
|
async def get_node_contract(request: Request, run_id: str, node_id: str):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
store = _store(request)
|
|
run = await _owned_run(store, run_id, user_id)
|
|
if isinstance(run, JSONResponse):
|
|
return run
|
|
try:
|
|
return await store.get_node_contract(run_id, node_id)
|
|
except Exception as exc:
|
|
return _map_store_error(exc)
|
|
|
|
|
|
@router.get("/runs/{run_id}/artifacts")
|
|
async def list_artifacts(request: Request, run_id: str):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
store = _store(request)
|
|
run = await _owned_run(store, run_id, user_id)
|
|
if isinstance(run, JSONResponse):
|
|
return run
|
|
return {"artifacts": await store.list_artifacts(run_id)}
|
|
|
|
|
|
@router.get("/runs/{run_id}/sources")
|
|
async def list_sources(request: Request, run_id: str):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
run = await _owned_run(_store(request), run_id, user_id)
|
|
if isinstance(run, JSONResponse):
|
|
return run
|
|
return {"sources": []}
|