feat(orchestration): config-gated delegate_jobs tool for cross-repo delegation

daiv/accounts/locale/pt/LC_MESSAGES/django.po

@@ -97,6 +97,9 @@ msgstr ""
 msgid "Your DAIV Sign-In Code"
 msgstr "O Seu Código de Autenticação DAIV"
 
+msgid "New chat"
+msgstr "Novo chat"
+
 msgid "Dashboard"
 msgstr "Painel"
 

daiv/automation/agent/base.py

@@ -17,10 +17,12 @@
 from core.constants import BOT_NAME
 from core.models import Provider, ProviderType
 from core.models import ThinkingLevelChoices as ThinkingLevel
+from core.site_settings import site_settings
 
 logger = logging.getLogger("daiv.automation")
 
 if TYPE_CHECKING:
+    import httpx
     from langchain_core.language_models.chat_models import BaseChatModel
     from langchain_core.messages import BaseMessage
     from langgraph.checkpoint.base import BaseCheckpointSaver
@@ -266,8 +268,44 @@ def _apply_insecure_http_clients(kw: dict, row: Provider.Cached) -> None:
         return
     import httpx
 
-    kw["http_client"] = httpx.Client(verify=False)  # noqa: S501  # admin-opted-in via Provider.verify_ssl
-    kw["http_async_client"] = httpx.AsyncClient(verify=False)  # noqa: S501
+    # Carry over the resolved request timeout; a custom client otherwise falls back to
+    # httpx's 5s default, which would cut off real generations.
+    timeout = kw["timeout"]
+    kw["http_client"] = httpx.Client(verify=False, timeout=timeout)  # noqa: S501  # admin-opted-in via Provider.verify_ssl
+    kw["http_async_client"] = httpx.AsyncClient(verify=False, timeout=timeout)  # noqa: S501
+
+
+def _resolve_request_timeout(provider_type: ProviderType, timeout_seconds: float) -> httpx.Timeout | float:
+    """Return a per-request timeout in the shape each provider's langchain integration accepts.
+
+    ``langchain-openai`` (OpenAI + OpenRouter) accepts an ``httpx.Timeout``, so we give the connect
+    phase its own short fuse while read/write/pool get the full budget. ``langchain-anthropic``
+    (``timeout: float | None``) and ``langchain-google-genai`` (scalar seconds, converted to ms
+    internally) only accept a plain number, so those get a bare float.
+    """
+    timeout_seconds = float(timeout_seconds)
+    if provider_type in _HTTPX_CLIENT_PROVIDER_TYPES:
+        import httpx
+
+        return httpx.Timeout(timeout_seconds, connect=min(10.0, timeout_seconds))
+    return timeout_seconds
+
+
+def _apply_request_timeout_and_retries(kw: dict, provider_type: ProviderType) -> None:
+    """Bound how long a model call can hang and how many times it retries.
+
+    Replaces the SDKs' own defaults (600s timeouts for OpenAI/Anthropic, unbounded for Google)
+    with the site-configured budget. Caller-supplied ``timeout`` / ``max_retries`` (e.g. the
+    tighter web_fetch budget) take precedence, but an explicit ``None`` is treated as unset so a
+    caller cannot silently reintroduce an unbounded default.
+    """
+    if kw.get("max_retries") is None:
+        retries = site_settings.model_max_retries
+        # langchain-google-genai's max_retries counts total *attempts* (0 and 1 both mean "no
+        # retries"), so translate to keep "N retries" meaning uniform across providers.
+        kw["max_retries"] = retries + 1 if provider_type == ProviderType.GOOGLE_GENAI else retries
+    if kw.get("timeout") is None:
+        kw["timeout"] = _resolve_request_timeout(provider_type, site_settings.model_request_timeout_seconds)
 
 
 _BARE_NAME_HEURISTICS = (
@@ -421,6 +459,8 @@ def get_model_kwargs(*, resolved: ResolvedProvider, thinking_level: ThinkingLeve
         else:
             raise RuntimeError(f"Unknown provider_type {row.provider_type!r} on slug {row.slug!r}")
 
+        _apply_request_timeout_and_retries(kw, row.provider_type)
+
         if not row.verify_ssl:
             _apply_insecure_http_clients(kw, row)
 

daiv/automation/agent/graph.py

@@ -26,6 +26,7 @@
 )
 from automation.agent.mcp.toolkits import MCPToolkit
 from automation.agent.middlewares.deferred_tools import deferred_tools_middleware, direct_mcp_tools
+from automation.agent.middlewares.delegate_jobs import DelegateJobsMiddleware
 from automation.agent.middlewares.ensure_response import ensure_non_empty_response
 from automation.agent.middlewares.file_system import (
     CUSTOM_TOOL_DESCRIPTIONS,
@@ -310,6 +311,7 @@ async def create_daiv_agent(
         # source of write_todos and the harness profile excludes nothing here.
         TodoListMiddleware(system_prompt=dynamic_write_todos_system_prompt(bash_tool_enabled=_sandbox_enabled)),
         *([SlashCommandMiddleware(subagents=subagents)] if ctx.config.slash_commands.enabled else []),
+        *([DelegateJobsMiddleware()] if ctx.config.orchestration.enabled else []),
         *(
             [SandboxMiddleware(agent_root=agent_root, client=run_client, sandbox_backend=sandbox_backend)]
             if _sandbox_enabled

daiv/automation/agent/middlewares/delegate_jobs.py

@@ -0,0 +1,190 @@
+from __future__ import annotations
+
+import json
+import logging
+from typing import TYPE_CHECKING
+
+from django.urls import reverse
+
+from langchain.agents.middleware import AgentMiddleware, ModelRequest, ModelResponse
+from langchain_core.runnables import RunnableConfig  # noqa: TCH002 — used in @tool signature at runtime
+from langchain_core.tools import tool
+from pydantic import BaseModel, Field
+from sandbox_envs.services import aresolve_repo_envs
+from sessions.models import MAX_SPAWN_DEPTH, Session, SessionOrigin
+from sessions.services import MAX_DELEGATED_TARGETS, RepoTarget, asubmit_batch_runs
+
+from codebase.authorization import REPO_ACCESS_DENIED_MESSAGE, RepositoryAccessDenied, aassert_can_run
+from codebase.conf import settings
+from codebase.models import RepositoryCatalog
+
+if TYPE_CHECKING:
+    from collections.abc import Awaitable, Callable
+
+logger = logging.getLogger("daiv.tools")
+
+DELEGATE_JOBS_NAME = "delegate_jobs"
+
+
+class DelegateTarget(BaseModel):
+    repo_id: str = Field(description="Identifier of the target repository.")
+    ref: str | None = Field(default=None, description="Base branch/ref to start from; omit for the default branch.")
+    prompt: str = Field(
+        min_length=1,
+        description="Self-contained instruction for this repository. The leg runs in an isolated session "
+        "and sees only this text — not your conversation, the originating request, or anything else you "
+        "can see here — so include all the context it needs. End with the no-change convention: "
+        "'if this repository is unaffected, reply saying so and make no changes.'",
+    )
+
+
+DELEGATE_JOBS_DESCRIPTION = f"""\
+Delegate tailored sub-jobs to other repositories and return immediately.
+
+Each target runs as an independent single-repo job (own thread, own MR). Your turn ends after
+delegating; you will be resumed automatically with a summary of every leg once they all finish.
+
+- `goal`: one-line description of the overall objective (used to title the batch).
+- `targets`: 1-{MAX_DELEGATED_TARGETS} entries, each `{{repo_id, ref?, prompt}}` — `prompt` required per target.
+
+Returns JSON:
+{{"batch_id", "delegated": [{{repo_id, ref, thread_id, session_url}}], "failed": [{{repo_id, error}}]}}."""
+
+
+DELEGATE_JOBS_SYSTEM_PROMPT = f"""\
+## Delegation tool `{DELEGATE_JOBS_NAME}`
+
+When a task spans other repositories, use `{DELEGATE_JOBS_NAME}` to fan tailored work out to them —
+each target runs as an independent job. This is only for work in *other* repositories; for parallel
+work inside this one, use the `task` tool (subagents) instead. If the task is contained to this
+repository, ignore this tool.
+After a call whose `delegated` list is non-empty, state your plan and end your turn — do NOT poll
+or wait. You will be resumed with the consolidated results when every delegated leg finishes.
+If `delegated` comes back empty, nothing was started and no resume will come — handle the failures
+in this same turn instead of waiting.
+Each leg runs in isolation and sees only the prompt you give it, so make every target's prompt
+self-contained — include the context it needs and the no-change convention.
+"""
+
+
+def _error(message: str) -> str:
+    return json.dumps({"error": message})
+
+
+@tool(DELEGATE_JOBS_NAME, description=DELEGATE_JOBS_DESCRIPTION)
+async def delegate_jobs_tool(goal: str, targets: list[DelegateTarget], config: RunnableConfig) -> str:
+    """Delegate per-repo sub-jobs; returns a JSON string (no state mutation)."""
+    thread_id = (config.get("configurable") or {}).get("thread_id")
+    if not thread_id:
+        return _error("delegate_jobs is only available inside a checkpointed run.")
+
+    session = await Session.objects.select_related("user").filter(thread_id=thread_id).afirst()
+    if session is None:
+        return _error("Could not resolve the current session for delegation.")
+    if session.user_id is None:
+        return _error("Delegation requires an authenticated coordinator; this session has no user.")
+
+    if not targets:
+        return _error("At least one target is required.")
+    if len(targets) > MAX_DELEGATED_TARGETS:
+        return _error(f"At most {MAX_DELEGATED_TARGETS} targets per delegate_jobs call.")
+    if session.spawn_depth >= MAX_SPAWN_DEPTH:
+        return _error(f"Delegation depth limit reached (MAX_SPAWN_DEPTH={MAX_SPAWN_DEPTH}).")
+    # An omitted ref and the default branch's explicit name are the same physical checkout, so
+    # resolve both through the synced catalog before comparing.
+    slugs = {session.repo_id} | {t.repo_id for t in targets}
+    default_branches: dict[str, str] = {
+        slug: branch
+        async for slug, branch in RepositoryCatalog.objects.filter(
+            provider=settings.CLIENT.value, slug__in=slugs
+        ).values_list("slug", "default_branch")
+    }
+
+    def checkout(repo_id: str, ref: str | None) -> tuple[str, str]:
+        return (repo_id, ref or default_branches.get(repo_id, ""))
+
+    coordinator_checkout = checkout(session.repo_id, session.ref)
+    if any(checkout(t.repo_id, t.ref) == coordinator_checkout for t in targets):
+        # In-repo parallelism belongs to subagents; a different ref on the same repo is a distinct
+        # checkout and may delegate.
+        return _error(
+            f"Cannot delegate to the coordinator's own checkout ({session.repo_id!r} on "
+            f"{session.ref or 'default branch'!r}). delegate_jobs fans work out to other "
+            "checkouts as independent jobs; for parallel work on this one, use the `task` "
+            "tool (subagents) instead."
+        )
+
+    seen: set[tuple[str, str]] = set()
+    for t in targets:
+        key = checkout(t.repo_id, t.ref)
+        if key in seen:
+            return _error(f"Duplicate target: {t.repo_id} on {t.ref or 'default branch'}.")
+        seen.add(key)
+
+    user = session.user
+
+    # Per-target authorization: aassert_can_run is all-or-nothing, so partition on the denied set.
+    denied: set[str] = set()
+    try:
+        await aassert_can_run(user, [t.repo_id for t in targets])
+    except RepositoryAccessDenied as exc:
+        denied = set(exc.repo_ids)
+
+    allowed = [t for t in targets if t.repo_id not in denied]
+    failed = [{"repo_id": rid, "error": REPO_ACCESS_DENIED_MESSAGE} for rid in sorted(denied)]
+
+    batch_id: str | None = None
+    delegated: list[dict] = []
+    if allowed:
+        repo_targets = [RepoTarget(repo_id=t.repo_id, ref=t.ref or "", prompt=t.prompt) for t in allowed]
+        # A raise here (OperationalError, revoked-access RepositoryAccessDenied, validation
+        # ValueError) must surface as the tool's JSON error contract, not a tool-node crash.
+        try:
+            repo_targets = await aresolve_repo_envs(user=user, repos=repo_targets, explicit_env_id=None)
+            result = await asubmit_batch_runs(
+                user=user,
+                prompt=goal,
+                repos=repo_targets,
+                trigger_type=SessionOrigin.DELEGATED_JOB,
+                parent_thread_id=thread_id,
+                spawn_depth=session.spawn_depth + 1,
+            )
+        except Exception:  # noqa: BLE001
+            logger.exception("delegate_jobs: submission failed for thread=%s", thread_id)
+            return _error("Delegation submission failed; no sub-jobs were started.")
+        batch_id = str(result.batch_id)
+        delegated = [
+            {
+                "repo_id": run.repo_id,
+                "ref": run.ref,
+                "thread_id": str(run.session_id),
+                "session_url": reverse("session_detail", kwargs={"thread_id": run.session_id}),
+            }
+            for run in result.runs
+        ]
+        failed.extend({"repo_id": f.repo_id, "error": f.error} for f in result.failed)
+
+    payload: dict = {"batch_id": batch_id, "delegated": delegated, "failed": failed}
+    if not delegated:
+        payload["note"] = (
+            "No sub-jobs are running. Do not end your turn to wait for a resume — none will come; "
+            "handle the failures now."
+        )
+    return json.dumps(payload, ensure_ascii=False)
+
+
+class DelegateJobsMiddleware(AgentMiddleware):
+    """Bind the delegate_jobs tool and inject its usage note. Added to the agent when
+    ``orchestration.enabled`` is set — on by default; a repo opts out with
+    ``orchestration.enabled: false``.
+    """
+
+    def __init__(self) -> None:
+        self.tools = [delegate_jobs_tool]
+
+    async def awrap_model_call(
+        self, request: ModelRequest, handler: Callable[[ModelRequest], Awaitable[ModelResponse]]
+    ) -> ModelResponse:
+        system_prompt = (request.system_prompt + "\n\n") if request.system_prompt else ""
+        system_prompt += DELEGATE_JOBS_SYSTEM_PROMPT
+        return await handler(request.override(system_prompt=system_prompt))

daiv/automation/agent/middlewares/web_fetch.py

@@ -27,6 +27,11 @@
 
 WEB_FETCH_NAME = "web_fetch"
 
+# The summariser runs a small/fast model over already-bounded page content
+# (``web_fetch_max_content_chars``), and the tool degrades to returning the raw
+# content on failure — so keep its budget tighter than the global model timeout.
+WEB_FETCH_MODEL_TIMEOUT_SECONDS = 60
+
 WEB_FETCH_TOOL_DESCRIPTION = """\
 Fetch content from a specified URL and process it using an AI model.
 
@@ -232,17 +237,30 @@ async def web_fetch_tool(
     if not prompt.strip() or site_settings.web_fetch_model_name is None:
         return f"Contents of {url}:\n{content}"
 
+    # Building the model is pure construction, so a failure here is a permanent provider
+    # misconfiguration that breaks every call — log at ERROR (Sentry) rather than degrade quietly.
+    try:
+        model = BaseAgent.get_model(model=site_settings.web_fetch_model_name, timeout=WEB_FETCH_MODEL_TIMEOUT_SECONDS)
+    except Exception as e:
+        logger.exception(
+            "web_fetch summariser model %r could not be built; check the provider configuration.",
+            site_settings.web_fetch_model_name,
+        )
+        return f"web_fetch summariser unavailable ({e}); returning raw content instead.\nContents of {url}:\n{content}"
+
+    messages = [
+        SystemMessage(
+            content=(
+                "You process web pages for users. Use the page content to answer the user's prompt.\n"
+                "Be concise. If the content doesn't contain the answer, say so."
+            )
+        ),
+        HumanMessage(content=f"URL: {url}\n\n<PageContent>\n{content}\n</PageContent>\n\nPrompt:\n{prompt}"),
+    ]
+
+    # Invocation failures are transient (timeout, network, provider error) — degrade gracefully
+    # to the raw page content so a blip doesn't abort the agent's turn.
     try:
-        model = BaseAgent.get_model(model=site_settings.web_fetch_model_name)
-        messages = [
-            SystemMessage(
-                content=(
-                    "You process web pages for users. Use the page content to answer the user's prompt.\n"
-                    "Be concise. If the content doesn't contain the answer, say so."
-                )
-            ),
-            HumanMessage(content=f"URL: {url}\n\n<PageContent>\n{content}\n</PageContent>\n\nPrompt:\n{prompt}"),
-        ]
         response = await model.ainvoke(messages)
         response_text = str(getattr(response, "content", response))
         _set_cached_response(url=url, prompt=prompt, response=response_text)

daiv/automation/agent/skills/orchestrate/SKILL.md

@@ -0,0 +1,43 @@
+---
+name: orchestrate
+description: Use when you are a coordinator on a coordination repo turning one request — a ticket, issue, security advisory, or direct instruction — into tailored per-repo sub-jobs via delegate_jobs, then reporting the consolidated outcome once every leg finishes.
+---
+
+# Orchestrate cross-repo work
+
+You run on a **coordination repository**. Your job is to turn one request — a ticket, issue,
+security advisory, or direct instruction — into tailored work across other repositories, then
+report the combined outcome.
+
+> **Precondition.** This skill depends on the `delegate_jobs` tool, which is bound by default but can
+> be disabled per-repository via `orchestration.enabled: false` in `.daiv.yml`. If `delegate_jobs` is
+> not in your available tools, orchestration has been disabled here — say so plainly and stop; do not
+> try to emulate delegation by other means.
+
+## Workflow
+
+1. **Triage.** Read the request using your attached MCP tools. Consult `AGENTS.md` in this repo for
+   the repository directory and routing rules. Decide which repositories are affected and what each
+   one must do.
+2. **Delegate.** Call `delegate_jobs(goal, targets)` once, with a tailored `prompt` per target.
+   - Each leg runs in an isolated session and sees only the prompt you give it — not this
+     conversation, the originating request, or your tool outputs. Put everything a leg needs
+     directly in its prompt.
+   - Include the no-change convention in each prompt: *"if this repository is unaffected, reply
+     saying so and make no changes."*
+3. **End your turn.** State your delegation plan and stop. Do **not** poll or wait — you will be
+   resumed automatically once every leg finishes.
+4. **Report.** On resume you receive a summary of all legs (status, MR links, replies). Verify the
+   outcome, compose the consolidated result, and report it back to wherever the request originated
+   (e.g. comment on the ticket or issue) using your MCP tools.
+5. **Follow up (optional).** If a sequenced change is needed (e.g. adapt repo B against repo A's
+   MR), delegate another batch — you will be resumed again.
+
+## Limits
+
+- Up to 10 targets per `delegate_jobs` call.
+- Delegation depth is capped; a leg cannot itself delegate beyond the configured chain depth.
+- You can only delegate as an authenticated coordinator; targets you lack write access to are
+  reported back to you as failures rather than run.
+
+Routing rules specific to your setup belong in this repo's `.agents/AGENTS.md`, not this skill.