feat: add in-session workflow nudges

ROADMAP.md

@@ -248,7 +248,7 @@ As usage grows, the platform needs stronger derived data pipelines, performance
 - [x] MCP sidecar with on-demand stats, friction reports, and recommendations
 - [x] [P0] Proactive coaching skill that activates at session start with contextual suggestions
 - [x] [P0] Live session signals that stream friction, satisfaction, and risk as work happens
-- [ ] [P1] In-session workflow nudges based on project playbooks and prior failures
+- [x] [P1] In-session workflow nudges based on project playbooks and prior failures
 - [ ] [P1] Daily and weekly personal recaps inside the sidecar
 - [ ] [P2] Lightweight session planning prompts before complex work begins
 

src/primer/common/schemas.py

@@ -2225,6 +2225,22 @@ class LiveSessionSignalsResponse(BaseModel):
     signals: list[LiveSessionSignal] = Field(default_factory=list)
 
 
+class InSessionNudge(BaseModel):
+    nudge_type: str
+    severity: Literal["info", "warning", "critical"]
+    title: str
+    message: str
+    rationale: str | None = None
+    suggested_actions: list[str] = Field(default_factory=list)
+
+
+class InSessionNudgesResponse(BaseModel):
+    session_id: str | None = None
+    project_name: str | None = None
+    risk_level: Literal["low", "medium", "high"]
+    nudges: list[InSessionNudge] = Field(default_factory=list)
+
+
 # --- Coaching Brief ---
 
 

src/primer/mcp/nudges.py

@@ -0,0 +1,161 @@
+from __future__ import annotations
+
+from primer.common.schemas import (
+    CoachingBrief,
+    InSessionNudge,
+    InSessionNudgesResponse,
+    LiveSessionSignal,
+    LiveSessionSignalsResponse,
+)
+
+
+def build_in_session_nudges(
+    live_signals: LiveSessionSignalsResponse,
+    coaching_brief: CoachingBrief | None = None,
+    *,
+    limit: int = 3,
+) -> InSessionNudgesResponse:
+    section_items = _section_items_by_title(coaching_brief)
+    nudges: list[InSessionNudge] = []
+
+    for signal in live_signals.signals:
+        nudge = _nudge_for_signal(signal, live_signals, section_items)
+        if nudge is None:
+            continue
+        nudges.append(nudge)
+
+    if not nudges and live_signals.risk_level == "low":
+        nudges.append(
+            InSessionNudge(
+                nudge_type="stay_the_course",
+                severity="info",
+                title="Stay on the current path",
+                message=(
+                    "No strong intervention signal is standing out right now. Keep the current "
+                    "workflow tight and only widen scope if the next verification step fails."
+                ),
+                rationale="Live signals look healthy and there is no urgent corrective action.",
+                suggested_actions=section_items.get("How to start this session", [])[:1],
+            )
+        )
+
+    nudges.sort(key=_nudge_sort_key)
+    return InSessionNudgesResponse(
+        session_id=live_signals.session_id,
+        project_name=live_signals.project_name,
+        risk_level=live_signals.risk_level,
+        nudges=nudges[:limit],
+    )
+
+
+def _section_items_by_title(coaching_brief: CoachingBrief | None) -> dict[str, list[str]]:
+    if coaching_brief is None:
+        return {}
+    return {section.title: list(section.items) for section in coaching_brief.sections}
+
+
+def _nudge_for_signal(
+    signal: LiveSessionSignal,
+    live_signals: LiveSessionSignalsResponse,
+    section_items: dict[str, list[str]],
+) -> InSessionNudge | None:
+    if signal.signal_type == "verification_failure":
+        return InSessionNudge(
+            nudge_type="tighten_verification_loop",
+            severity=signal.severity,
+            title="Tighten the verification loop",
+            message=(
+                "Recent verification steps are failing. Narrow the loop to the smallest proving "
+                "command before making broader edits."
+            ),
+            rationale=signal.detail,
+            suggested_actions=_merge_actions(
+                section_items.get("How to start this session", []),
+                section_items.get("What to reach for", []),
+            )[:2],
+        )
+
+    if signal.signal_type == "tool_error_cluster":
+        return InSessionNudge(
+            nudge_type="fallback_from_tooling",
+            severity=signal.severity,
+            title="Switch away from the failing tool path",
+            message=(
+                "Multiple tool errors are stacking up. Move to the project fallback path or a "
+                "simpler local inspection step instead of retrying the same integration."
+            ),
+            rationale=signal.detail,
+            suggested_actions=_merge_actions(
+                section_items.get("What to watch for", []),
+                section_items.get("How to start this session", []),
+            )[:2],
+        )
+
+    if signal.signal_type == "recovery_loop":
+        return InSessionNudge(
+            nudge_type="break_recovery_loop",
+            severity=signal.severity,
+            title="Break the recovery loop",
+            message=(
+                "The session looks stuck in repeated recovery attempts. Pause, return to the best "
+                "known playbook, and pick a single next proving action."
+            ),
+            rationale=signal.detail,
+            suggested_actions=_merge_actions(
+                section_items.get("How to start this session", []),
+                section_items.get("What to reach for", []),
+            )[:3],
+        )
+
+    if signal.signal_type == "negative_sentiment":
+        return InSessionNudge(
+            nudge_type="reframe_task",
+            severity=signal.severity,
+            title="Reframe before the next step",
+            message=(
+                "Recent user messages look frustrated. Reset to the smallest next step that can "
+                "confirm or disprove the current hypothesis."
+            ),
+            rationale=signal.detail,
+            suggested_actions=_merge_actions(
+                section_items.get("How to start this session", []),
+                section_items.get("What to watch for", []),
+            )[:2],
+        )
+
+    if signal.signal_type == "positive_momentum" and live_signals.risk_level == "low":
+        return InSessionNudge(
+            nudge_type="maintain_momentum",
+            severity="info",
+            title="Keep the loop tight",
+            message=(
+                "The session looks healthy. Stay with the current workflow and avoid adding extra "
+                "scope until the next verification step."
+            ),
+            rationale=signal.detail,
+            suggested_actions=section_items.get("How to start this session", [])[:1],
+        )
+
+    return None
+
+
+def _merge_actions(*groups: list[str]) -> list[str]:
+    merged: list[str] = []
+    seen: set[str] = set()
+    for group in groups:
+        for item in group:
+            normalized = item.strip()
+            if not normalized or normalized in seen:
+                continue
+            seen.add(normalized)
+            merged.append(normalized)
+    return merged
+
+
+def _nudge_sort_key(nudge: InSessionNudge) -> tuple[int, int, str]:
+    severity_score = {"critical": 2, "warning": 1, "info": 0}
+    return (
+        -severity_score.get(nudge.severity, 0),
+        -len(nudge.suggested_actions),
+        nudge.title,
+    )

src/primer/mcp/server.py

@@ -8,6 +8,7 @@
 from primer.mcp.tools import (
     primer_coaching,
     primer_friction_report,
+    primer_in_session_nudges,
     primer_live_session_signals,
     primer_my_stats,
     primer_recommendations,
@@ -107,5 +108,27 @@ def live_session_signals(
     return primer_live_session_signals(session_id=session_id, transcript_path=transcript_path)
 
 
+@mcp.tool()
+def in_session_nudges(
+    project_name: str | None = None,
+    workflow_hint: str | None = None,
+    task_hint: str | None = None,
+    session_id: str | None = None,
+    transcript_path: str | None = None,
+) -> str:
+    """Get evidence-backed nudges for what to try next during an active session.
+
+    Primer combines live local session signals with project playbooks and prior
+    team evidence to suggest the smallest next corrective action.
+    """
+    return primer_in_session_nudges(
+        project_name=project_name,
+        workflow_hint=workflow_hint,
+        task_hint=task_hint,
+        session_id=session_id,
+        transcript_path=transcript_path,
+    )
+
+
 if __name__ == "__main__":
     mcp.run()

src/primer/mcp/tools.py

@@ -5,7 +5,9 @@
 import os
 
 import httpx
+from pydantic import ValidationError
 
+from primer.mcp.nudges import build_in_session_nudges
 from primer.mcp.sync import sync_sessions
 from primer.server.services.live_session_signal_service import get_live_session_signals
 
@@ -205,3 +207,68 @@ def primer_live_session_signals(
         lines.append(f"- {signal.detail}")
         lines.append("")
     return "\n".join(lines)
+
+
+def primer_in_session_nudges(
+    project_name: str | None = None,
+    workflow_hint: str | None = None,
+    task_hint: str | None = None,
+    session_id: str | None = None,
+    transcript_path: str | None = None,
+) -> str:
+    """Get evidence-backed nudges during an active local session.
+
+    Combines local live-session signals with the session-start coaching brief so the
+    sidecar can suggest the smallest next corrective action when the session drifts.
+    """
+    try:
+        live_signals = get_live_session_signals(
+            session_id=session_id,
+            transcript_path=transcript_path,
+        )
+    except ValueError as exc:
+        return f"Error: {exc}"
+
+    coaching_brief = None
+    if API_KEY:
+        try:
+            params = {
+                key: value
+                for key, value in {
+                    "project_name": project_name or live_signals.project_name,
+                    "workflow_hint": workflow_hint,
+                    "task_hint": task_hint,
+                    "days": 90,
+                }.items()
+                if value is not None
+            }
+            resp = httpx.get(
+                f"{SERVER_URL}/api/v1/analytics/coaching/session-start",
+                params=params,
+                headers={"x-api-key": API_KEY},
+                timeout=30,
+            )
+            if resp.status_code == 200:
+                from primer.common.schemas import CoachingBrief
+
+                coaching_brief = CoachingBrief.model_validate(resp.json())
+        except (ValidationError, ValueError, httpx.RequestError, json.JSONDecodeError):
+            coaching_brief = None
+
+    data = build_in_session_nudges(live_signals, coaching_brief)
+    lines = ["## In-Session Workflow Nudges\n"]
+    lines.append(f"**Risk**: {data.risk_level}\n")
+    if data.project_name:
+        lines.append(f"**Project**: {data.project_name}\n")
+    if not data.nudges:
+        lines.append("No nudges right now.\n")
+        return "\n".join(lines)
+    for nudge in data.nudges:
+        lines.append(f"### [{nudge.severity}] {nudge.title}")
+        lines.append(f"- {nudge.message}")
+        if nudge.rationale:
+            lines.append(f"- Why now: {nudge.rationale}")
+        for action in nudge.suggested_actions:
+            lines.append(f"- Try: {action}")
+        lines.append("")
+    return "\n".join(lines)

tests/test_mcp_server.py

@@ -89,3 +89,25 @@ def test_live_session_signals_tool(mock_live_signals):
         session_id="session-123",
         transcript_path=None,
     )
+
+
+@patch("primer.mcp.server.primer_in_session_nudges")
+def test_in_session_nudges_tool(mock_nudges):
+    mock_nudges.return_value = "nudges"
+
+    from primer.mcp.server import in_session_nudges
+
+    result = in_session_nudges(
+        project_name="api-server",
+        workflow_hint="debugging",
+        task_hint="Fix auth regression",
+        session_id="session-123",
+    )
+    assert result == "nudges"
+    mock_nudges.assert_called_once_with(
+        project_name="api-server",
+        workflow_hint="debugging",
+        task_hint="Fix auth regression",
+        session_id="session-123",
+        transcript_path=None,
+    )