Merge pull request #121 from MLAI-AUS-Inc/codex/domain-scan-routing-fix

adding publish as PR skill

Author: drsamdonegan

roo-standalone/roo/agent.py

@@ -146,16 +146,21 @@ def remember_thread_context(
         *,
         domain: Optional[str] = None,
         workflow: Optional[str] = None,
+        active_job_id: Optional[str] = None,
     ) -> None:
         """Persist recent thread routing context so follow-ups stay on the right skill."""
         thread_key = self._thread_key(channel_id, thread_ts)
         if not thread_key or not skill_name:
             return
 
+        existing = self._thread_skill_context.get(thread_key, {})
         self._thread_skill_context[thread_key] = {
-            "skill_name": skill_name,
-            "domain": domain,
-            "workflow": workflow,
+            "skill_name": skill_name or existing.get("skill_name"),
+            "domain": domain if domain is not None else existing.get("domain"),
+            "workflow": workflow if workflow is not None else existing.get("workflow"),
+            "active_job_id": (
+                active_job_id if active_job_id is not None else existing.get("active_job_id")
+            ),
             "updated_at": datetime.now(timezone.utc),
         }
 
@@ -227,6 +232,7 @@ def _get_routing_intent(
             text,
             thread_skill_name=(thread_context or {}).get("skill_name"),
             thread_domain=(thread_context or {}).get("domain"),
+            thread_job_id=(thread_context or {}).get("active_job_id"),
         )
         if not route:
             return None

roo-standalone/roo/clients/mlai_backend.py

@@ -945,6 +945,23 @@ async def publish_article(self, job_id: str, slack_user_id: str) -> dict:
             response.raise_for_status()
             return response.json()
 
+    async def publish_article_as_pr(self, job_id: str, slack_user_id: str) -> dict:
+        """Promote a completed content-only article into a draft-PR publish run."""
+        if not self.base_url:
+            raise ValueError("MLAI_BACKEND_URL not configured")
+
+        clean_id = self._clean_slack_id(slack_user_id)
+
+        async with httpx.AsyncClient() as client:
+            response = await client.post(
+                f"{self.base_url}/api/v1/content/jobs/{job_id}/publish-pr",
+                json={"slack_user_id": clean_id},
+                headers=self.headers,
+                timeout=60.0,
+            )
+            response.raise_for_status()
+            return response.json()
+
     # =========================================================================
     # Missing Admin / Points Methods
     # =========================================================================

roo-standalone/roo/content_intent.py

@@ -32,6 +32,12 @@
     r"\bwhat\s+should\s+i\s+write\b",
     r"\b(?:recommend|suggest)\s+(?:a\s+)?(?:topic|article|keyword)\b",
 )
+PUBLISH_PR_PATTERNS = (
+    r"\bpublish\b.*\b(?:article|bundle|draft|post)\b.*\bas\s+a\s+p\.?r\.?\b",
+    r"\bpublish\b.*\bas\s+a\s+pull\s+request\b",
+    r"\bturn\b.*\b(?:article|bundle|draft|post)\b.*\binto\s+a\s+p\.?r\.?\b",
+    r"\bopen\b.*\b(?:a\s+)?(?:draft\s+)?p\.?r\.?\b",
+)
 WRITE_PATTERNS = (
     r"\bwrite\b.*\b(article|blog(?:\s+post)?|content)\b",
     r"\bgenerate\b.*\b(article|blog(?:\s+post)?|content)\b",
@@ -89,6 +95,8 @@ def detect_content_action(text: str) -> Optional[str]:
         return "scaffold"
     if any(re.search(pattern, text_lower) for pattern in RESEARCH_PATTERNS):
         return "research"
+    if any(re.search(pattern, text_lower) for pattern in PUBLISH_PR_PATTERNS):
+        return "publish_pr"
     if any(re.search(pattern, text_lower) for pattern in WRITE_PATTERNS):
         return "write"
     return None
@@ -108,6 +116,7 @@ def parse_routing_intent(
     *,
     thread_skill_name: Optional[str] = None,
     thread_domain: Optional[str] = None,
+    thread_job_id: Optional[str] = None,
 ) -> Optional[Dict[str, Any]]:
     """Return a deterministic routing decision for common content flows."""
     normalized = normalize_slack_text(text)
@@ -150,6 +159,18 @@ def parse_routing_intent(
             params["domain"] = domain
         return {"skill_name": "content-factory", "params": params}
 
+    if (
+        action == "publish_pr"
+        and thread_skill_name == "content-factory"
+        and (thread_job_id or thread_domain)
+    ):
+        params = {"action": "publish_pr"}
+        if thread_job_id:
+            params["job_id"] = thread_job_id
+        if domain:
+            params["domain"] = domain
+        return {"skill_name": "content-factory", "params": params}
+
     if action == "write" and (
         domain or re.search(r"\b(?:article|blog(?:\s+post)?|content)\b", text_lower)
     ):

roo-standalone/roo/main.py

@@ -228,6 +228,13 @@ async def _maybe_attach_content_factory_progress(
     if not message_ts:
         return
 
+    workflow = str(
+        result_data.get("content_factory_workflow")
+        or result_data.get("content_factory_watchdog_mode")
+        or ""
+    ).strip() or None
+    domain = str(result_data.get("content_factory_domain") or "").strip() or None
+
     from .clients.mlai_backend import MLAIBackendClient
 
     settings = get_settings()
@@ -248,6 +255,14 @@ async def _maybe_attach_content_factory_progress(
         print(f"⚠️ Failed to attach progress message {message_ts} to job {job_id}: {exc}")
         return
 
+    _remember_content_thread_context(
+        channel_id,
+        thread_ts,
+        domain,
+        workflow or "write",
+        active_job_id=job_id,
+    )
+
     if result_data.get("content_factory_watchdog"):
         asyncio.create_task(_watch_content_factory_quiet_run(job_id))
 
@@ -323,6 +338,13 @@ async def _trigger_article_generation_from_pending(
         )
         print(f"✅ Auto-generation triggered for {domain}")
         if result.get("job_id") or result.get("run_id"):
+            _remember_content_thread_context(
+                intent_channel,
+                intent_thread,
+                domain,
+                "research" if include_decision_stage else "write",
+                active_job_id=result.get("job_id") or result.get("run_id"),
+            )
             asyncio.create_task(_watch_content_factory_quiet_run(result.get("job_id") or result.get("run_id")))
         return True
     except Exception as e:
@@ -341,6 +363,8 @@ def _remember_content_thread_context(
     thread_ts: str | None,
     domain: str | None,
     workflow: str,
+    *,
+    active_job_id: str | None = None,
 ) -> None:
     """Keep content-factory as the active skill for follow-ups in this thread."""
     if not channel_id or not thread_ts:
@@ -353,6 +377,7 @@ def _remember_content_thread_context(
             thread_ts,
             domain=domain,
             workflow=workflow,
+            active_job_id=active_job_id,
         )
     except Exception as e:
         print(f"⚠️ Failed to persist content thread context: {e}")
@@ -2508,6 +2533,13 @@ async def slack_actions(request: Request):
             )
             print(f"✅ Article generation triggered for {domain}: {result}")
             if result.get("job_id") or result.get("run_id"):
+                _remember_content_thread_context(
+                    reply_channel,
+                    reply_thread_ts,
+                    domain,
+                    "write",
+                    active_job_id=result.get("job_id") or result.get("run_id"),
+                )
                 asyncio.create_task(_watch_content_factory_quiet_run(result.get("job_id") or result.get("run_id")))
         except Exception as e:
             print(f"❌ Failed to trigger article generation: {e}")
@@ -2636,6 +2668,13 @@ async def slack_actions(request: Request):
             )
             print(f"✅ Article discovery triggered for {domain}: {result}")
             if str(result.get("status") or "").strip().lower() == "awaiting_delivery_mode":
+                _remember_content_thread_context(
+                    reply_channel,
+                    reply_thread_ts,
+                    domain,
+                    "awaiting_delivery_mode",
+                    active_job_id=result.get("job_id") or result.get("run_id"),
+                )
                 try:
                     from .slack_client import get_slack_client
                     get_slack_client().chat_update(
@@ -2651,6 +2690,13 @@ async def slack_actions(request: Request):
                 except Exception as update_error:
                     print(f"⚠️ Failed to update delivery-mode prompt: {update_error}")
             elif result.get("job_id") or result.get("run_id"):
+                _remember_content_thread_context(
+                    reply_channel,
+                    reply_thread_ts,
+                    domain,
+                    "research",
+                    active_job_id=result.get("job_id") or result.get("run_id"),
+                )
                 asyncio.create_task(_watch_content_factory_quiet_run(result.get("job_id") or result.get("run_id")))
         except Exception as e:
             print(f"❌ Failed to trigger article discovery: {e}")
@@ -2761,6 +2807,13 @@ async def slack_actions(request: Request):
                 request_source=CONTENT_FACTORY_REQUEST_SOURCE,
             )
             if result.get("job_id") or result.get("run_id"):
+                _remember_content_thread_context(
+                    payload.get("channel", {}).get("id"),
+                    payload.get("message", {}).get("thread_ts") or payload.get("message", {}).get("ts"),
+                    domain,
+                    "awaiting_delivery_mode",
+                    active_job_id=result.get("job_id") or result.get("run_id"),
+                )
                 asyncio.create_task(_watch_content_factory_quiet_run(result.get("job_id") or result.get("run_id")))
 
             selected_label = "content-only" if delivery_mode == "content_only" else "publish-via-code"
@@ -3322,6 +3375,13 @@ async def slack_actions(request: Request):
             )
             print(f"✅ Topic confirmed for job {job_id}")
             follow_up = _build_confirm_topic_follow_up(result)
+            _remember_content_thread_context(
+                payload.get("channel", {}).get("id"),
+                payload.get("message", {}).get("thread_ts") or payload.get("message", {}).get("ts"),
+                follow_up.get("domain"),
+                "awaiting_delivery_mode" if follow_up.get("requires_delivery_mode") else "write",
+                active_job_id=follow_up.get("active_job_id"),
+            )
 
             try:
                 from .slack_client import get_slack_client

roo-standalone/roo/skills/executor.py

@@ -386,6 +386,8 @@ def _build_article_delivery_mode_prompt(
                 "content_factory_progress_job_id": job_id,
                 "content_factory_watchdog": False,
                 "content_factory_watchdog_mode": "awaiting_delivery_mode",
+                "content_factory_domain": domain,
+                "content_factory_workflow": "awaiting_delivery_mode",
             },
         }
 
@@ -1335,6 +1337,38 @@ def _build_content_factory_start_response(
                 "content_factory_progress_job_id": job_id,
                 "content_factory_watchdog": True,
                 "content_factory_watchdog_mode": workflow,
+                "content_factory_domain": domain,
+                "content_factory_workflow": workflow,
+            },
+        }
+
+    def _build_publish_pr_start_response(
+        self,
+        *,
+        domain: Optional[str],
+        job_id: str,
+    ) -> dict:
+        display_domain = normalize_content_factory_domain(domain) or domain or "this domain"
+        return {
+            "message": (
+                f"Publishing the completed article bundle for {display_domain} as a draft PR. "
+                "I'll keep this message updated as the run moves forward."
+            ),
+            "blocks": build_live_status_blocks(
+                display_domain,
+                summary_text=(
+                    "Promoting the completed article bundle into the repo and preview flow. "
+                    "I'll keep this message updated."
+                ),
+                include_decision_stage=False,
+                current_stage="preparing",
+            ),
+            "data": {
+                "content_factory_progress_job_id": job_id,
+                "content_factory_watchdog": True,
+                "content_factory_watchdog_mode": "publish_pr",
+                "content_factory_domain": display_domain,
+                "content_factory_workflow": "publish_pr",
             },
         }
 
@@ -1352,6 +1386,56 @@ def _get_connected_domain_info(
             None,
         )
 
+    async def _publish_content_bundle_as_pr(
+        self,
+        api_client,
+        *,
+        job_id: str,
+        domain: Optional[str],
+        slack_user_id: str,
+    ) -> Any:
+        if not job_id:
+            return (
+                "I couldn't tell which article bundle to publish as a PR in this thread. "
+                "Please ask from the original article thread after the content-ready message."
+            )
+
+        try:
+            response = await api_client.publish_article_as_pr(job_id, slack_user_id)
+        except httpx.HTTPStatusError as exc:
+            error_message = str(exc)
+            try:
+                error_data = exc.response.json()
+            except Exception:
+                error_data = {}
+            if isinstance(error_data, dict):
+                error_message = (
+                    error_data.get("error")
+                    or error_data.get("message")
+                    or error_message
+                )
+            return f"❌ I couldn't publish that article bundle as a PR: {error_message}"
+        except Exception as exc:
+            return f"❌ I couldn't publish that article bundle as a PR: {exc}"
+
+        child_job_id = str(response.get("job_id") or response.get("run_id") or "").strip()
+        if not child_job_id:
+            return (
+                "I asked Content Factory to publish that article bundle as a PR, "
+                "but it didn't return a run ID."
+            )
+
+        resolved_domain = (
+            normalize_content_factory_domain(domain)
+            or normalize_content_factory_domain(response.get("domain"))
+            or domain
+            or response.get("domain")
+        )
+        return self._build_publish_pr_start_response(
+            domain=resolved_domain,
+            job_id=child_job_id,
+        )
+
     def _resolve_content_factory_repo_name(
         self,
         integration: dict,
@@ -1395,6 +1479,13 @@ async def _execute_content_factory(
         domain = params.get("domain")
         org_config_cached = None
         action = params.get("action")
+        if action == "publish_pr":
+            return await self._publish_content_bundle_as_pr(
+                api_client,
+                job_id=str(params.get("job_id") or "").strip(),
+                domain=domain,
+                slack_user_id=user_id,
+            )
         is_scan_request = self._is_explicit_scan_request(text, params)
         is_article_flow = action != "scaffold" and not is_scan_request
         requested_delivery_mode, requested_delivery_mode_confirmed = self._resolve_requested_article_delivery_mode(