fix: agent autonomous execution + project-context recovery

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "product-manager-agent"
-version = "0.3.11"
+version = "0.3.12"
 description = "Stateful Technical PM Agent — multi-project CLI with persistent per-project context"
 readme = "README.md"
 requires-python = ">=3.12"

src/pm_agent/__init__.py

@@ -2,6 +2,6 @@
 
 from .config import AgentConfig
 
-__version__ = "0.3.7"
+__version__ = "0.3.12"
 
 __all__ = ["AgentConfig", "__version__"]

src/pm_agent/application/conversation_service.py

@@ -3,6 +3,7 @@
 import logging
 from dataclasses import replace
 
+from pm_agent.application.decision_policy import DecisionPolicy
 from pm_agent.domain.enums import ActionStatus, DecisionStatus
 from pm_agent.domain.models import (
     PMResponse,
@@ -100,6 +101,17 @@ def handle(
                     f"Repair error: {repair_error}"
                 ) from repair_error
 
+        # Make the agent's ask-vs-act stance explicit and surface regressions
+        # (e.g. delegating work it could do itself) as risks rather than asking.
+        needs = DecisionPolicy.classify(response)
+        warnings = DecisionPolicy.validate(response)
+        if warnings:
+            response = replace(
+                response, execution_needs=needs, risks=[*response.risks, *warnings]
+            )
+        else:
+            response = replace(response, execution_needs=needs)
+
         stored_actions = []
         approved_candidates = []
         blocked_operations: list[str] = []
@@ -158,7 +170,16 @@ def handle(
                     reason,
                     candidate.payload,
                 )
-                blocked_operations.append(f"{proposal.operation} (reason: {reason})")
+                category = DecisionPolicy.categorize_block(proposal.operation)
+                if category == "external_access":
+                    blocked_operations.append(
+                        f"{proposal.operation} (blocked: missing external access/permission "
+                        f"- {reason})"
+                    )
+                else:
+                    blocked_operations.append(
+                        f"{proposal.operation} (blocked: needs approval - {reason})"
+                    )
 
         if blocked_operations or len(approved_candidates) != len(response.actions_requiring_approval):
             response = replace(

src/pm_agent/application/decision_policy.py

@@ -0,0 +1,98 @@
+from __future__ import annotations
+
+import re
+
+from pm_agent.domain.enums import TaskClass
+from pm_agent.domain.models import ExecutionNeeds, PMResponse
+
+# Phrases that indicate the model is pushing executable work back to the user
+# instead of doing it. These are heuristics used only to surface a warning, not
+# to block the response.
+_OFFLOAD_PATTERNS = [
+    re.compile(
+        r"you should (inspect|review|read|check|look at|examine) (the|your) "
+        r"(repo|repository|codebase|project)",
+        re.I,
+    ),
+    re.compile(
+        r"(please |kindly |manually )?(inspect|review|write|create|document|"
+        r"break down|analyze) (the|your) (repo|repository|docs?|documentation|"
+        r"issues?|breakdown|mvp)",
+        re.I,
+    ),
+    re.compile(r"(you need to|you must) (write|create|inspect|review|document|break down)", re.I),
+]
+
+
+class DecisionPolicy:
+    """Lightweight guardrail around the model's ask-vs-act decision.
+
+    The authoritative behaviour lives in ``prompts/system.md`` (autonomy and
+    clarification policy). This class provides a code-side safety net so the
+    agent's stance is explicit, testable, and can be flagged when it regresses.
+    """
+
+    @staticmethod
+    def classify(response: PMResponse) -> ExecutionNeeds:
+        """Return the explicit execution needs, deriving a default if absent."""
+        if response.execution_needs is not None:
+            return response.execution_needs
+        if response.actions_requiring_approval:
+            classification = TaskClass.AGENT_EXECUTABLE
+        elif response.decisions:
+            classification = TaskClass.USER_DECISION_REQUIRED
+        else:
+            classification = TaskClass.AGENT_EXECUTABLE
+        return ExecutionNeeds(classification=classification)
+
+    @staticmethod
+    def detect_offloading(response: PMResponse) -> list[str]:
+        """Warn when the response delegates agent-executable work to the user."""
+        haystacks = [response.summary, response.analysis, *response.recommendations]
+        for decision in response.decisions:
+            haystacks.append(f"{decision.title} {decision.decision} {decision.reason}")
+        text = "\n".join(haystacks)
+        for pattern in _OFFLOAD_PATTERNS:
+            if pattern.search(text):
+                return [
+                    "Response delegates work to the user that the agent can perform "
+                    "autonomously from accessible artifacts (repo, issues, memory). "
+                    "Prefer emitting actions/analysis over asking the user to do it."
+                ]
+        return []
+
+    @staticmethod
+    def validate(response: PMResponse) -> list[str]:
+        """Return human-readable warnings about the ask-vs-act stance."""
+        warnings: list[str] = []
+        needs = DecisionPolicy.classify(response)
+        warnings.extend(DecisionPolicy.detect_offloading(response))
+        if needs.classification is TaskClass.USER_DECISION_REQUIRED and not needs.open_questions:
+            warnings.append(
+                "Classified as user_decision_required but no open_questions were "
+                "supplied. Narrow the missing decision to a single explicit question."
+            )
+        if needs.classification is TaskClass.EXTERNAL_ACCESS_REQUIRED and not needs.missing_access:
+            warnings.append(
+                "Classified as external_access_required but missing_access is empty. "
+                "Name exactly what permission or integration access is missing."
+            )
+        return warnings
+
+    @staticmethod
+    def categorize_block(operation: str) -> str:
+        """Categorize a blocked action so refusal messaging is precise."""
+        if operation.startswith("github") or operation in {
+            "create_issue",
+            "create_issues",
+            "create_issue_comment",
+            "create_sub_issue",
+            "create_milestone",
+            "update_milestone",
+            "setup_sprint",
+            "add_issue_to_project",
+            "create_project",
+            "update_issue",
+        }:
+            return "external_access"
+        return "approval"

src/pm_agent/domain/actions.py

@@ -47,6 +47,12 @@ class GitHubAction:
     "create_issue": GitHubAction("create_issue", "write", frozenset({Capability.WRITE_ISSUES})),
     "update_issue": GitHubAction("update_issue", "write", frozenset({Capability.WRITE_ISSUES})),
     "create_issues": GitHubAction("create_issues", "write", frozenset({Capability.WRITE_ISSUES})),
+    "create_issue_comment": GitHubAction(
+        "create_issue_comment", "write", frozenset({Capability.WRITE_ISSUES})
+    ),
+    "create_sub_issue": GitHubAction(
+        "create_sub_issue", "write", frozenset({Capability.WRITE_ISSUES})
+    ),
     "setup_sprint": GitHubAction("setup_sprint", "write", frozenset({Capability.WRITE_MILESTONES})),
     "add_issue_to_project": GitHubAction("add_issue_to_project", "write", frozenset({Capability.WRITE_PROJECTS})),
     "create_project": GitHubAction("create_project", "write", frozenset({Capability.WRITE_PROJECTS})),

src/pm_agent/domain/enums.py

@@ -34,3 +34,17 @@ class MemoryKind(StrEnum):
     REPO_NOTE = "repo_note"
     ACTION_OUTCOME = "action_outcome"
     MESSAGE = "message"
+
+
+class TaskClass(StrEnum):
+    """Classification of a task the agent is asked to perform.
+
+    Used to decide whether the agent should act autonomously, ask the user a
+    narrow question, or report a missing external capability. See
+    ``pm_agent.application.decision_policy`` for how it is applied.
+    """
+
+    AGENT_EXECUTABLE = "agent_executable"
+    AGENT_EXECUTABLE_WITH_ASSUMPTIONS = "agent_executable_with_assumptions"
+    USER_DECISION_REQUIRED = "user_decision_required"
+    EXTERNAL_ACCESS_REQUIRED = "external_access_required"

src/pm_agent/domain/models.py

@@ -8,7 +8,7 @@
 from typing import Any
 from uuid import uuid4
 
-from .enums import ActionStatus, ActionType, DecisionStatus, MemoryKind
+from .enums import ActionStatus, ActionType, DecisionStatus, MemoryKind, TaskClass
 
 
 def new_id() -> str:
@@ -221,6 +221,21 @@ class DecisionCandidate:
     status: DecisionStatus = DecisionStatus.PROPOSED
 
 
+@dataclass(frozen=True)
+class ExecutionNeeds:
+    """How the agent should proceed with a task, surfaced in the model response.
+
+    ``classification`` is one of :class:`TaskClass`. The remaining fields make
+    the agent's stance explicit so the REPL and tests can verify the agent is
+    not offloading work it could do itself.
+    """
+
+    classification: TaskClass
+    assumptions: list[str] = field(default_factory=list)
+    open_questions: list[str] = field(default_factory=list)
+    missing_access: list[str] = field(default_factory=list)
+
+
 @dataclass(frozen=True)
 class PMResponse:
     summary: str
@@ -229,6 +244,7 @@ class PMResponse:
     recommendations: list[str] = field(default_factory=list)
     decisions: list[DecisionCandidate] = field(default_factory=list)
     actions_requiring_approval: list[ActionCandidate] = field(default_factory=list)
+    execution_needs: ExecutionNeeds | None = None
 
 
 @dataclass(frozen=True)

src/pm_agent/domain/policies.py

@@ -265,6 +265,16 @@ def populated_string(value: Any) -> bool:
                     "add_issue_to_project requires positive issue_numbers and a project "
                     "number or title."
                 )
+        elif operation == "create_issue_comment":
+            if not isinstance(payload.get("issue_number"), int) or payload["issue_number"] <= 0:
+                return "create_issue_comment requires a positive integer issue_number."
+            if not populated_string(payload.get("body")):
+                return "create_issue_comment requires an exact comment body."
+        elif operation == "create_sub_issue":
+            if not isinstance(payload.get("parent"), int) or payload["parent"] <= 0:
+                return "create_sub_issue requires a positive integer parent issue number."
+            if not populated_string(payload.get("title")):
+                return "create_sub_issue requires an exact sub-issue title."
         return None
 
     def _evaluate_mcp(
@@ -275,6 +285,16 @@ def _evaluate_mcp(
             "filesystem", "git", "memory", "graphify", "sequential_thinking", "github"
         }:
             return PolicyDecision(False, "blocked", "Unknown MCP category.")
+        if operation == "write_document":
+            if "path" not in payload or "content" not in payload:
+                return PolicyDecision(
+                    False,
+                    "blocked",
+                    "write_document requires 'path' and 'content' in payload.",
+                )
+            return PolicyDecision(
+                True, "medium", "Sandboxed document write (repo-relative) requires approval."
+            )
         if normalized in {"filesystem", "git", "github"} and any(
             word in operation for word in _WRITE_OPERATIONS
         ):