feat(goal): add research goal runtime

Add the session-scoped Research Goal layer across backend, CLI, API/MCP,
SSE, and Web UI. Goals now persist claims, acceptance criteria, evidence,
budgets, and completion policy; agent tools and /goal can create goals and
attach evidence; REST/MCP expose snapshots and evidence writes; SSE keeps
chat clients fresh.

Also carry the 2026-05-24 transplant cleanup: replay-all/session-id/upload
wiring, report-worthy Full Report gating, English-only Web UI copy, and audit
fixes for verified evidence, live-trading risk tiers, CLI session linkage,
goal ledger cleanup, replay-all hookup, and frontend snapshot races.

Author: warren618

agent/api_server.py

@@ -238,6 +238,63 @@ class MessageResponse(BaseModel):
     metadata: Optional[Dict[str, Any]] = None
 
 
+class CreateGoalRequest(BaseModel):
+    """Create or replace a finance research goal."""
+
+    objective: str = Field(..., min_length=1, max_length=5000)
+    criteria: List[str] = Field(default_factory=list)
+    ui_summary: str = ""
+    protocol: str = "thesis_review"
+    risk_tier: str = "research_general"
+    token_budget: Optional[int] = Field(None, ge=1)
+    turn_budget: Optional[int] = Field(None, ge=1)
+    time_budget_seconds: Optional[int] = Field(None, ge=1)
+
+
+class AddGoalEvidenceRequest(BaseModel):
+    """Append evidence to a finance research goal."""
+
+    goal_id: str = Field(..., min_length=1)
+    expected_goal_id: str = Field(..., min_length=1)
+    text: str = Field(..., min_length=1, max_length=10000)
+    criterion_id: Optional[str] = None
+    claim_id: Optional[str] = None
+    evidence_type: str = "evidence"
+    tool_call_id: Optional[str] = None
+    run_id: Optional[str] = None
+    source_provider: Optional[str] = None
+    source_type: Optional[str] = None
+    source_uri: Optional[str] = None
+    symbol_universe: List[str] = Field(default_factory=list)
+    benchmark: List[str] = Field(default_factory=list)
+    timeframe: Optional[str] = None
+    method: Optional[str] = None
+    assumptions: Dict[str, Any] = Field(default_factory=dict)
+    artifact_path: Optional[str] = None
+    artifact_hash: Optional[str] = None
+    data_as_of: Optional[str] = None
+    confidence: Optional[str] = None
+    caveat: Optional[str] = None
+    contradicts_claim_ids: List[str] = Field(default_factory=list)
+
+
+class GoalSnapshotResponse(BaseModel):
+    """Finance research goal snapshot."""
+
+    goal: Dict[str, Any]
+    claims: List[Dict[str, Any]]
+    criteria: List[Dict[str, Any]]
+    evidence: List[Dict[str, Any]]
+    evidence_count: int = 0
+
+
+class AddGoalEvidenceResponse(BaseModel):
+    """Response after appending goal evidence."""
+
+    evidence: Dict[str, Any]
+    snapshot: GoalSnapshotResponse
+
+
 
 # ============================================================================
 # FastAPI Application
@@ -1320,6 +1377,7 @@ async def api_info():
 # ============================================================================
 
 _session_service = None
+_goal_store = None
 
 
 def _get_session_service():
@@ -1353,6 +1411,26 @@ def _get_session_service():
     return _session_service
 
 
+def _get_goal_store():
+    """Return the shared finance goal store."""
+    global _goal_store
+    if _goal_store is None:
+        from src.goal import GoalStore
+
+        _goal_store = GoalStore()
+    return _goal_store
+
+
+def _get_existing_session_or_404(session_id: str):
+    svc = _get_session_service()
+    if not svc:
+        raise HTTPException(status_code=501, detail="Session runtime not enabled")
+    session = svc.get_session(session_id)
+    if not session:
+        raise HTTPException(status_code=404, detail=f"Session {session_id} not found")
+    return svc, session
+
+
 @app.post("/sessions", response_model=SessionResponse, status_code=status.HTTP_201_CREATED, dependencies=[Depends(require_auth)])
 async def create_session(request: CreateSessionRequest):
     """Create a chat session."""
@@ -1410,6 +1488,124 @@ async def get_session(session_id: str):
     )
 
 
+@app.post(
+    "/sessions/{session_id}/goal",
+    response_model=GoalSnapshotResponse,
+    status_code=status.HTTP_201_CREATED,
+    dependencies=[Depends(require_auth)],
+)
+async def create_session_goal(session_id: str, req: CreateGoalRequest):
+    """Create or replace the current finance research goal for a session."""
+    _validate_path_param(session_id, "session_id")
+    svc, _session = _get_existing_session_or_404(session_id)
+    from src.goal import RiskTier
+
+    criteria = [item.strip() for item in req.criteria if item.strip()]
+    if not criteria:
+        criteria = ["Define research thesis", "Record at least one supporting or contradicting evidence row"]
+    try:
+        risk_tier = RiskTier(req.risk_tier)
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=f"invalid risk_tier: {req.risk_tier}") from exc
+    if risk_tier is RiskTier.LIVE_TRADING_OR_EXECUTION:
+        raise HTTPException(status_code=400, detail="live trading or execution goals are not supported")
+
+    goal_store = _get_goal_store()
+    try:
+        goal = goal_store.replace_goal(
+            session_id=session_id,
+            objective=req.objective,
+            criteria=criteria,
+            ui_summary=req.ui_summary,
+            source="api",
+            protocol=req.protocol,
+            risk_tier=risk_tier,
+            token_budget=req.token_budget,
+            turn_budget=req.turn_budget,
+            time_budget_seconds=req.time_budget_seconds,
+        )
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+    snapshot = goal_store.get_goal_snapshot(goal.goal_id)
+    if snapshot is None:
+        raise HTTPException(status_code=500, detail="Goal created but could not be reloaded")
+    svc.event_bus.emit(session_id, "goal.created", {"goal": snapshot["goal"]})
+    return snapshot
+
+
+@app.get(
+    "/sessions/{session_id}/goal",
+    response_model=GoalSnapshotResponse,
+    dependencies=[Depends(require_auth)],
+)
+async def get_session_goal(session_id: str):
+    """Return the current finance research goal snapshot for a session."""
+    _validate_path_param(session_id, "session_id")
+    _get_existing_session_or_404(session_id)
+    snapshot = _get_goal_store().get_current_snapshot(session_id)
+    if snapshot is None:
+        raise HTTPException(status_code=404, detail="No current goal")
+    return snapshot
+
+
+@app.post(
+    "/sessions/{session_id}/goal/evidence",
+    response_model=AddGoalEvidenceResponse,
+    status_code=status.HTTP_201_CREATED,
+    dependencies=[Depends(require_auth)],
+)
+async def add_session_goal_evidence(session_id: str, req: AddGoalEvidenceRequest):
+    """Append traceable evidence to the current finance research goal."""
+    _validate_path_param(session_id, "session_id")
+    svc, _session = _get_existing_session_or_404(session_id)
+    from dataclasses import asdict
+    from src.goal import EvidenceInput, StaleGoalError
+
+    goal_store = _get_goal_store()
+    try:
+        evidence = goal_store.append_evidence(
+            session_id=session_id,
+            goal_id=req.goal_id,
+            expected_goal_id=req.expected_goal_id,
+            evidence=EvidenceInput(
+                criterion_id=req.criterion_id,
+                claim_id=req.claim_id,
+                evidence_type=req.evidence_type,
+                text=req.text,
+                tool_call_id=req.tool_call_id,
+                run_id=req.run_id,
+                source_provider=req.source_provider,
+                source_type=req.source_type,
+                source_uri=req.source_uri,
+                symbol_universe=req.symbol_universe,
+                benchmark=req.benchmark,
+                timeframe=req.timeframe,
+                method=req.method,
+                assumptions=req.assumptions,
+                artifact_path=req.artifact_path,
+                artifact_hash=req.artifact_hash,
+                data_as_of=req.data_as_of,
+                confidence=req.confidence,
+                caveat=req.caveat,
+                contradicts_claim_ids=req.contradicts_claim_ids,
+            ),
+        )
+    except StaleGoalError as exc:
+        raise HTTPException(status_code=409, detail=str(exc)) from exc
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+    snapshot = goal_store.get_goal_snapshot(req.goal_id)
+    if snapshot is None:
+        raise HTTPException(status_code=500, detail="Goal snapshot could not be reloaded")
+    svc.event_bus.emit(
+        session_id,
+        "goal.evidence",
+        {"evidence": asdict(evidence), "goal_id": req.goal_id},
+    )
+    return {"evidence": asdict(evidence), "snapshot": snapshot}
+
+
 @app.delete("/sessions/{session_id}", dependencies=[Depends(require_auth)])
 async def delete_session(session_id: str):
     """Delete a session."""
@@ -1420,6 +1616,7 @@ async def delete_session(session_id: str):
     deleted = svc.delete_session(session_id)
     if not deleted:
         raise HTTPException(status_code=404, detail=f"Session {session_id} not found")
+    _get_goal_store().delete_session_goals(session_id)
     return {"status": "deleted", "session_id": session_id}
 
 
@@ -1504,6 +1701,7 @@ async def session_events(
     session_id: str,
     request: Request,
     last_event_id: Optional[str] = Query(None, alias="Last-Event-ID"),
+    replay: Optional[str] = Query(None),
 ):
     """SSE stream for agent events."""
     _validate_path_param(session_id, "session_id")
@@ -1516,9 +1714,19 @@ async def session_events(
 
     header_id = request.headers.get("Last-Event-ID")
     event_id = header_id or last_event_id
+    replay_active = (replay or "").lower() == "active"
+    replay_all = False
+    if replay_active and not event_id and session.last_attempt_id:
+        attempt = svc.store.get_attempt(session_id, session.last_attempt_id)
+        attempt_status = getattr(attempt.status, "value", attempt.status) if attempt else None
+        replay_all = attempt_status == "running"
 
     async def event_generator():
-        async for event in svc.event_bus.subscribe(session_id, last_event_id=event_id):
+        async for event in svc.event_bus.subscribe(
+            session_id,
+            last_event_id=event_id,
+            replay_all=replay_all,
+        ):
             if await request.is_disconnected():
                 break
             yield event.to_sse()
@@ -1594,20 +1802,19 @@ async def upload_file(file: UploadFile):
     if not file.filename:
         raise HTTPException(status_code=400, detail="Missing filename")
     filename = Path(file.filename).name
-    ext = Path(file.filename).suffix.lower()
+    ext = Path(filename).suffix.lower()
     if ext in _BLOCKED_UPLOAD_EXT or filename.lower() in _BLOCKED_UPLOAD_NAMES:
         raise HTTPException(
             status_code=400,
             detail="This file type is not allowed for upload.",
         )
 
-    UPLOADS_DIR.mkdir(parents=True, exist_ok=True)
-
     safe_name = f"{uuid.uuid4().hex}{ext}"
     dest = UPLOADS_DIR / safe_name
     total_size = 0
 
     try:
+        UPLOADS_DIR.mkdir(parents=True, exist_ok=True)
         with dest.open("wb") as handle:
             while True:
                 chunk = await file.read(_UPLOAD_CHUNK_SIZE)
@@ -1628,14 +1835,17 @@ async def upload_file(file: UploadFile):
     except OSError as exc:
         if dest.exists():
             dest.unlink()
-        raise HTTPException(status_code=500, detail=f"Failed to store upload: {exc}") from exc
+        raise HTTPException(
+            status_code=500,
+            detail="Upload failed while storing the file. Please retry or choose a different file.",
+        ) from exc
     finally:
         await file.close()
 
     return {
         "status": "ok",
-        "file_path": str(dest.resolve()),
-        "filename": file.filename,
+        "file_path": f"uploads/{safe_name}",
+        "filename": filename,
     }
 
 

agent/cli/_legacy.py

@@ -850,6 +850,7 @@ def _run_agent(
     no_rich: bool = False,
     stream_output: bool = True,
     dashboard: Optional[_RunDashboard] = None,
+    session_id: str = "",
 ) -> dict:
     """Build AgentLoop and execute, return result dict."""
     from src.tools import build_registry
@@ -975,6 +976,7 @@ def _mcp_warn(msg: str) -> None:
             persistent_memory=pm,
             include_shell_tools=True,
             agent_config=agent_config,
+            session_id=session_id or None,
             warn_callback=_mcp_warn,
         ),
         llm=ChatLLM(),
@@ -985,7 +987,13 @@ def _mcp_warn(msg: str) -> None:
     if run_dir_override:
         agent.memory.run_dir = run_dir_override
 
-    return _run_with_graceful_cancel(agent, prompt, history, no_rich=no_rich)
+    return _run_with_graceful_cancel(
+        agent,
+        prompt,
+        history,
+        no_rich=no_rich,
+        session_id=session_id,
+    )
 
 
 def _run_with_graceful_cancel(
@@ -994,6 +1002,7 @@ def _run_with_graceful_cancel(
     history: Optional[List[Dict]],
     *,
     no_rich: bool,
+    session_id: str = "",
 ) -> dict:
     """Run an agent loop with first-Ctrl+C = graceful cancel.
 
@@ -1020,7 +1029,7 @@ def _run_with_graceful_cancel(
         original = _signal.getsignal(_signal.SIGINT)
     except (ValueError, AttributeError):
         # Not on a thread that can receive signals — skip the handler swap.
-        return agent.run(user_message=prompt, history=history)
+        return agent.run(user_message=prompt, history=history, session_id=session_id)
 
     def _on_sigint(_signum, _frame) -> None:
         now = time.time()
@@ -1041,10 +1050,10 @@ def _on_sigint(_signum, _frame) -> None:
         _signal.signal(_signal.SIGINT, _on_sigint)
     except (ValueError, OSError):
         # signal.signal only works on the main thread of the main interpreter.
-        return agent.run(user_message=prompt, history=history)
+        return agent.run(user_message=prompt, history=history, session_id=session_id)
 
     try:
-        return agent.run(user_message=prompt, history=history)
+        return agent.run(user_message=prompt, history=history, session_id=session_id)
     finally:
         try:
             _signal.signal(_signal.SIGINT, original)