diff --git a/.env.example b/.env.example
index 1d62b5cb6..0c643c431 100644
--- a/.env.example
+++ b/.env.example
@@ -108,6 +108,11 @@ CONTAINER_LLM_MODEL=
 # Increase if you see truncated specs/PRDs (default 4096 is often too small)
 LLM_MAX_TOKENS=16384
 
+# Maximum LangGraph steps for each host ForgeAgent invocation (default 100).
+# A positive forge.agent_recursion_limit Jira project property overrides this.
+# Changes to this environment value require a worker restart.
+AGENT_RECURSION_LIMIT=100
+
 # =============================================================================
 # MCP Servers Configuration
 # =============================================================================
diff --git a/docs/reference/config.md b/docs/reference/config.md
index 5f875946a..a7bba6817 100644
--- a/docs/reference/config.md
+++ b/docs/reference/config.md
@@ -343,6 +343,44 @@ repositories:
     change_request_mode: direct
 ```
 
+## Host agent recursion limit
+
+`AGENT_RECURSION_LIMIT` sets the positive global maximum number of LangGraph
+steps for each host `ForgeAgent` invocation; it defaults to `100`. A Jira
+project can override it with a positive JSON integer in
+`forge.agent_recursion_limit`:
+
+```bash
+forge project-setup MYPROJ --agent-recursion-limit 150
+forge get-config MYPROJ --property forge.agent_recursion_limit
+forge project-setup MYPROJ --clear-agent-recursion-limit
+```
+
+An absent property or JSON `null` uses the global value, even when
+`FORGE_REQUIRE_PROJECT_CONFIG=true`. Other JSON types, zero, and negative
+numbers are invalid. Forge reads the property once at the start of each
+ticket-scoped host agent run. Temporary Jira read failures receive bounded
+retries, then stop the run with an error; Forge does not assume the global
+value after a failed read. Runs without a ticket use the global value without
+contacting Jira. The limit applies separately to each graph invocation,
+including a permitted structured-output fallback or transient retry. It does
+not cap wall-clock time or the entire ticket across workflow retries.
+
+`forge get-config MYPROJ --json` shows the raw project property, this CLI
+process's global fallback, and the effective value and source. A malformed
+property or failed read appears as an error with no claimed effective value.
+The CLI may have a different environment from the running worker. Project
+changes affect the next host agent run; changing the worker's environment
+value requires a worker restart. Container implementation agents use their
+own execution limits.
+
+Full configuration output and `--property forge.agent_recursion_limit` fail with
+a nonzero exit code if the recursion setting is malformed or cannot be read.
+`--models` and queries for unrelated properties (for example,
+`--property forge.repos`) do not read or validate the recursion setting, so its
+errors and retries cannot block those queries. This also applies to the
+`project-config` alias; host agent execution remains fail-closed.
+
 ## Proposal review configuration
 
 Projects can opt into GitHub pull-request review for PRDs and specifications.
diff --git a/src/forge/agent_recursion_policy.py b/src/forge/agent_recursion_policy.py
new file mode 100644
index 000000000..b384f282d
--- /dev/null
+++ b/src/forge/agent_recursion_policy.py
@@ -0,0 +1,142 @@
+"""Resolve the host-agent graph limit from global and Jira project settings."""
+
+import asyncio
+import logging
+import math
+import time
+from datetime import UTC, datetime
+from email.utils import parsedate_to_datetime
+from typing import TYPE_CHECKING, Any
+
+import httpx
+
+if TYPE_CHECKING:
+    from forge.config import Settings
+    from forge.integrations.jira.client import JiraClient
+
+
+AGENT_RECURSION_PROPERTY = "forge.agent_recursion_limit"
+MAX_READ_ATTEMPTS = 3
+READ_ATTEMPT_TIMEOUT = 10.0
+READ_TOTAL_BUDGET = 30.0
+READ_BACKOFF = (0.5, 1.0)
+logger = logging.getLogger(__name__)
+
+
+def validate_project_agent_recursion_limit(raw: Any, project_key: str) -> int | None:
+    """Validate one optional JSON project override without applying fallback."""
+    if raw is None:
+        return None
+    if type(raw) is not int or raw < 1:
+        raise ValueError(f"{AGENT_RECURSION_PROPERTY} for {project_key} must be a positive integer")
+    return raw
+
+
+def _monotonic() -> float:
+    return time.monotonic()
+
+
+async def _sleep(delay: float) -> None:
+    await asyncio.sleep(delay)
+
+
+async def _await_with_timeout(awaitable: Any, timeout: float) -> Any:
+    return await asyncio.wait_for(awaitable, timeout=timeout)
+
+
+def _retry_after_seconds(value: str | None) -> float | None:
+    if value is None:
+        return None
+    try:
+        delay = float(value)
+    except ValueError:
+        try:
+            retry_date = parsedate_to_datetime(value)
+            if retry_date.tzinfo is None:
+                retry_date = retry_date.replace(tzinfo=UTC)
+            delay = max(0.0, (retry_date - datetime.now(UTC)).total_seconds())
+        except (TypeError, ValueError):
+            return None
+    return delay if math.isfinite(delay) and delay >= 0 else None
+
+
+async def read_agent_recursion_property(jira: "JiraClient", project_key: str) -> Any | None:
+    """Read one project property with bounded transport retries only."""
+    deadline = _monotonic() + READ_TOTAL_BUDGET
+    for attempt in range(1, MAX_READ_ATTEMPTS + 1):
+        remaining = deadline - _monotonic()
+        if remaining <= 0:
+            raise RuntimeError(
+                f"Reading {AGENT_RECURSION_PROPERTY} for {project_key} exceeded the time budget"
+            )
+        retry_after = None
+        try:
+            return await _await_with_timeout(
+                jira.get_project_property(project_key, AGENT_RECURSION_PROPERTY),
+                min(READ_ATTEMPT_TIMEOUT, remaining),
+            )
+        except asyncio.CancelledError:
+            raise
+        except httpx.HTTPStatusError as error:
+            status = error.response.status_code
+            if status == 404:
+                return None
+            if status != 429 and status < 500:
+                raise RuntimeError(
+                    f"Reading {AGENT_RECURSION_PROPERTY} for {project_key} failed: HTTP {status}"
+                ) from None
+            reason = f"HTTP {status}"
+            retry_after = _retry_after_seconds(error.response.headers.get("Retry-After"))
+        except (httpx.TransportError, TimeoutError) as error:
+            reason = "timeout" if isinstance(error, TimeoutError) else "transport error"
+        except (ValueError, KeyError, TypeError):
+            raise ValueError(
+                f"Malformed {AGENT_RECURSION_PROPERTY} response for {project_key}"
+            ) from None
+
+        if attempt == MAX_READ_ATTEMPTS:
+            raise RuntimeError(
+                f"Reading {AGENT_RECURSION_PROPERTY} for {project_key} failed "
+                f"after {attempt} attempts: {reason}"
+            ) from None
+        remaining = deadline - _monotonic()
+        if remaining <= 0:
+            raise RuntimeError(
+                f"Reading {AGENT_RECURSION_PROPERTY} for {project_key} exceeded the time budget"
+            ) from None
+        delay = retry_after if retry_after is not None else READ_BACKOFF[attempt - 1]
+        if delay >= remaining:
+            raise RuntimeError(
+                f"Reading {AGENT_RECURSION_PROPERTY} for {project_key} exceeded the retry budget"
+            ) from None
+        await _sleep(delay)
+    raise AssertionError("unreachable")
+
+
+async def resolve_agent_recursion_limit_for_project(
+    settings: "Settings",
+    project_key: str | None,
+    *,
+    jira: "JiraClient | None" = None,
+) -> int:
+    """Resolve one host invocation's effective graph limit."""
+    owned_jira = None
+    if not project_key:
+        limit = settings.agent_recursion_limit
+        source = "global"
+    else:
+        if jira is None:
+            from forge.integrations.jira.client import JiraClient
+
+            owned_jira = JiraClient(settings=settings)
+            jira = owned_jira
+        try:
+            raw = await read_agent_recursion_property(jira, project_key)
+            override = validate_project_agent_recursion_limit(raw, project_key)
+        finally:
+            if owned_jira is not None:
+                await owned_jira.close()
+        limit = settings.agent_recursion_limit if override is None else override
+        source = "global" if override is None else "project"
+    logger.info("Host agent recursion limit=%s source=%s project=%s", limit, source, project_key)
+    return limit
diff --git a/src/forge/cli.py b/src/forge/cli.py
index 3efea284b..3f7b046ff 100644
--- a/src/forge/cli.py
+++ b/src/forge/cli.py
@@ -532,10 +532,25 @@ async def cmd_project_setup(args: argparse.Namespace) -> int:
     import json
     import sys
 
+    from forge.agent_recursion_policy import validate_project_agent_recursion_limit
     from forge.integrations.jira.client import JiraClient
 
     project_key = args.project_key.upper()
     is_json = getattr(args, "json", False)
+    agent_recursion_limit = getattr(args, "agent_recursion_limit", None)
+    clear_agent_recursion_limit = getattr(args, "clear_agent_recursion_limit", False)
+    if agent_recursion_limit is not None and clear_agent_recursion_limit:
+        print(
+            "Error: --agent-recursion-limit cannot be combined with --clear-agent-recursion-limit",
+            file=sys.stderr,
+        )
+        return 1
+    if agent_recursion_limit is not None:
+        try:
+            validate_project_agent_recursion_limit(agent_recursion_limit, project_key)
+        except ValueError as error:
+            print(f"Error: {error}", file=sys.stderr)
+            return 1
 
     jira = None
     original_stdout = sys.stdout
@@ -953,6 +968,21 @@ def repo_name(repo: str | dict) -> str | None:
                 for ref in current_references:
                     desc_str = f" - {ref.get('description')}" if ref.get("description") else ""
                     print(f"  {ref.get('url')}{desc_str}")
+        if clear_agent_recursion_limit:
+            await jira.delete_project_property(project_key, "forge.agent_recursion_limit")
+            mutations["forge.agent_recursion_limit"] = {"operation": "remove", "value": None}
+            if not is_json:
+                print("[OK] forge.agent_recursion_limit removed")
+        elif agent_recursion_limit is not None:
+            await jira.set_project_property(
+                project_key, "forge.agent_recursion_limit", agent_recursion_limit
+            )
+            mutations["forge.agent_recursion_limit"] = {
+                "operation": "set",
+                "value": agent_recursion_limit,
+            }
+            if not is_json:
+                print(f"[OK] forge.agent_recursion_limit = {agent_recursion_limit}")
 
         if not any(
             [
@@ -977,6 +1007,8 @@ def repo_name(repo: str | dict) -> str | None:
                 add_reference,
                 remove_reference,
                 list_references,
+                agent_recursion_limit is not None,
+                clear_agent_recursion_limit,
             ]
         ):
             msg = (
@@ -989,6 +1021,7 @@ def repo_name(repo: str | dict) -> str | None:
                 ", --remove-model, --clear-model-policy"
                 ", --clear-model-default"
                 ", --add-reference, --remove-reference, --list-references"
+                ", --agent-recursion-limit, --clear-agent-recursion-limit"
             )
             if is_json:
                 print(msg, file=sys.stderr)
@@ -1027,11 +1060,19 @@ async def cmd_get_config(args: argparse.Namespace) -> int:
 
     import httpx
 
+    from forge.agent_recursion_policy import (
+        AGENT_RECURSION_PROPERTY,
+        read_agent_recursion_property,
+        validate_project_agent_recursion_limit,
+    )
     from forge.config import get_settings
     from forge.integrations.jira.client import JiraClient
 
     project_key = args.project_key.upper()
     settings = get_settings()
+    include_recursion = not getattr(args, "models", False) and (
+        not args.property or args.property.lower() == AGENT_RECURSION_PROPERTY
+    )
 
     jira = JiraClient(settings=settings)
     try:
@@ -1059,15 +1100,33 @@ async def cmd_get_config(args: argparse.Namespace) -> int:
             "forge.references",
             "forge.model_policy",
             "forge.model_default",
+            AGENT_RECURSION_PROPERTY,
         ]
 
         # Combine standard keys with extra discovered keys (avoid duplicates, preserve order/sort)
         extra_keys = sorted(set(forge_discovered_keys) - set(standard_keys))
-        all_keys = standard_keys + extra_keys
+        all_keys = [
+            key
+            for key in standard_keys + extra_keys
+            if include_recursion or key != AGENT_RECURSION_PROPERTY
+        ]
 
         # Retrieve raw property values from Jira
         project_properties_raw = {}
+        recursion_error: str | None = None
         for key in all_keys:
+            if key == AGENT_RECURSION_PROPERTY:
+                try:
+                    project_properties_raw[key] = await read_agent_recursion_property(
+                        jira, project_key
+                    )
+                except (RuntimeError, ValueError) as error:
+                    recursion_error = str(error)
+                    project_properties_raw[key] = None
+                except Exception:
+                    recursion_error = f"Reading {AGENT_RECURSION_PROPERTY} for {project_key} failed"
+                    project_properties_raw[key] = None
+                continue
             try:
                 val = await jira.get_project_property(project_key, key)
                 project_properties_raw[key] = val
@@ -1081,6 +1140,9 @@ async def cmd_get_config(args: argparse.Namespace) -> int:
         # Clean/type-validate project properties
         project_properties = {}
         for key, val in project_properties_raw.items():
+            if key == AGENT_RECURSION_PROPERTY:
+                project_properties[key] = val
+                continue
             if val is None:
                 project_properties[key] = None
                 continue
@@ -1208,6 +1270,31 @@ async def cmd_get_config(args: argparse.Namespace) -> int:
             "source": "project" if model_default_val is not None else "unset",
         }
 
+        if include_recursion:
+            recursion_raw = project_properties[AGENT_RECURSION_PROPERTY]
+            if recursion_error is None:
+                try:
+                    recursion_override = validate_project_agent_recursion_limit(
+                        recursion_raw, project_key
+                    )
+                except ValueError as error:
+                    recursion_error = str(error)
+            if recursion_error is None:
+                effective_config[AGENT_RECURSION_PROPERTY] = {
+                    "value": (
+                        settings.agent_recursion_limit
+                        if recursion_override is None
+                        else recursion_override
+                    ),
+                    "source": "global" if recursion_override is None else "project",
+                }
+            else:
+                effective_config[AGENT_RECURSION_PROPERTY] = {
+                    "value": None,
+                    "source": "error",
+                    "error": recursion_error,
+                }
+
         if getattr(args, "models", False):
             try:
                 resolved = settings.model_policy_resolver().resolve_all(
@@ -1235,6 +1322,14 @@ async def cmd_get_config(args: argparse.Namespace) -> int:
                 print(f"Error: Unknown property '{args.property}'", file=sys.stderr)
                 return 1
             canonical_key = effective_keys_lower[query_key]
+            if recursion_error:
+                if canonical_key == AGENT_RECURSION_PROPERTY and recursion_raw is not None:
+                    print(
+                        f"Raw {AGENT_RECURSION_PROPERTY}: {json.dumps(recursion_raw)}",
+                        file=sys.stderr,
+                    )
+                print(f"Error: {recursion_error}", file=sys.stderr)
+                return 1
             value = effective_config[canonical_key]["value"]
 
             if value is None:
@@ -1257,6 +1352,7 @@ async def cmd_get_config(args: argparse.Namespace) -> int:
                 "PRD_PROPOSALS_PATH": settings.prd_proposals_path.strip("/")
                 if settings.prd_proposals_path
                 else None,
+                "AGENT_RECURSION_LIMIT": settings.agent_recursion_limit,
             }
             output_data = {
                 "project": project_key,
@@ -1265,6 +1361,9 @@ async def cmd_get_config(args: argparse.Namespace) -> int:
                 "effective": effective_config,
             }
             print(json.dumps(output_data, indent=2))
+            if recursion_error:
+                print(f"Error: {recursion_error}", file=sys.stderr)
+                return 1
             return 0
 
         # Output - Human-Readable (Default)
@@ -1295,6 +1394,7 @@ def fmt_fallback(v: Any) -> str:
         print(f"  {'GITHUB_DEFAULT_REPO:':<29} {fmt_fallback(settings.github_default_repo)}")
         print(f"  {'PRD_PROPOSALS_REPO:':<29} {fmt_fallback(settings.prd_proposals_repo)}")
         print(f"  {'PRD_PROPOSALS_PATH:':<29} {fmt_fallback(settings.prd_proposals_path)}")
+        print(f"  {'AGENT_RECURSION_LIMIT:':<29} {settings.agent_recursion_limit}")
 
         print("\nEffective configuration:")
         for key in all_keys:
@@ -1308,9 +1408,14 @@ def fmt_fallback(v: Any) -> str:
                 print(f"  {key + ':':<27} (none) [unset]")
             elif src == "global":
                 print(f"  {key + ':':<27} {fmt_val(val)} [default]")
+            elif src == "error":
+                print(f"  {key + ':':<27} (error) [{cfg['error']}]")
             else:
                 print(f"  {key + ':':<27} {fmt_val(val)} [{src}]")
 
+        if recursion_error:
+            print(f"Error: {recursion_error}", file=sys.stderr)
+            return 1
         return 0
 
     finally:
@@ -1917,6 +2022,18 @@ def main(argv: list[str] | None = None) -> int:
 """,
     )
     setup_parser.add_argument("project_key", help="Jira project key (e.g., MYPROJ)")
+    recursion_group = setup_parser.add_mutually_exclusive_group()
+    recursion_group.add_argument(
+        "--agent-recursion-limit",
+        type=int,
+        metavar="STEPS",
+        help="Override the host agent graph limit with a positive integer",
+    )
+    recursion_group.add_argument(
+        "--clear-agent-recursion-limit",
+        action="store_true",
+        help="Remove the project host agent graph limit override",
+    )
     setup_parser.add_argument(
         "--repo",
         action="append",
diff --git a/src/forge/config.py b/src/forge/config.py
index ac2acb24f..d2a240546 100644
--- a/src/forge/config.py
+++ b/src/forge/config.py
@@ -195,6 +195,11 @@ def known_repos(self) -> list[str]:
         default=16384,
         description="Maximum output tokens for LLM responses (default 16384)",
     )
+    agent_recursion_limit: int = Field(
+        default=100,
+        gt=0,
+        description="Maximum graph steps for each host agent invocation",
+    )
     model_connections: dict[str, Any] = Field(
         default_factory=dict,
         description="Named non-secret model connections (JSON object)",
diff --git a/src/forge/integrations/agents/agent.py b/src/forge/integrations/agents/agent.py
index c826bca3f..ddca6c7eb 100644
--- a/src/forge/integrations/agents/agent.py
+++ b/src/forge/integrations/agents/agent.py
@@ -20,6 +20,7 @@
 from langchain.agents.structured_output import ProviderStrategy, ToolStrategy
 from langchain_anthropic import ChatAnthropic
 from langgraph.checkpoint.memory import MemorySaver
+from langgraph.errors import GraphRecursionError
 from pydantic import BaseModel
 
 # Optional MCP support
@@ -32,6 +33,7 @@
     StructuredTool = None  # type: ignore[misc, assignment]
     HAS_MCP = False
 
+from forge.agent_recursion_policy import resolve_agent_recursion_limit_for_project
 from forge.config import Settings, get_settings
 from forge.integrations.agents.structured_outputs import ArtifactDocument, EpicDecomposition
 from forge.integrations.langfuse import get_langfuse_config, get_langfuse_context
@@ -661,6 +663,7 @@ async def _run_agent(
         metadata: dict[str, Any] | None = None,
         model_target: ResolvedModelTarget | None = None,
         response_schema: type[StructuredResponseT] | None = None,
+        recursion_limit: int | None = None,
     ) -> str | StructuredResponseT:
         """Run the agent with the given prompt.
 
@@ -717,6 +720,10 @@ async def _run_agent(
             config.update(langfuse_config)
         else:
             langfuse_ctx_params = {}
+        config["configurable"] = {**config.get("configurable", {}), "thread_id": thread_id}
+        config["recursion_limit"] = (
+            recursion_limit if recursion_limit is not None else self.settings.agent_recursion_limit
+        )
 
         # Invoke the agent with retry logic for transient errors
         # Use async Langfuse context for session tracking (v3+ API)
@@ -747,6 +754,14 @@ async def _run_agent(
                             result["structured_response"]
                         )
                     break  # Success, exit retry loop
+                except GraphRecursionError:
+                    logger.error(
+                        "Host agent graph exhausted: task=%s ticket=%s recursion_limit=%s",
+                        trace_name,
+                        ticket_key,
+                        config["recursion_limit"],
+                    )
+                    raise
                 except Exception as e:
                     last_error = e
                     if vertex_claude and not (
@@ -939,6 +954,9 @@ async def run_task(
             project_key,
             runtime_policy_key,
         )
+        recursion_limit = await resolve_agent_recursion_limit_for_project(
+            self.settings, project_key
+        )
         trace_state["llm_model"] = (
             model_target.model if model_target is not None else self.settings.llm_model
         )
@@ -955,6 +973,7 @@ async def run_task(
             metadata=trace_metadata or None,
             model_target=model_target,
             response_schema=response_schema,
+            recursion_limit=recursion_limit,
         )
         observe_agent_duration(task_type=task, duration=time.monotonic() - _start)
 
diff --git a/src/forge/models/artifacts.py b/src/forge/models/artifacts.py
index b4bbe4a95..536a9bd29 100644
--- a/src/forge/models/artifacts.py
+++ b/src/forge/models/artifacts.py
@@ -7,6 +7,8 @@
 from dataclasses import dataclass, field
 from datetime import datetime
 
+from forge.utils.time import utc_now_naive
+
 
 @dataclass
 class Feature:
@@ -17,8 +19,8 @@ class Feature:
     prd_content: str = ""
     spec_content: str = ""
     epic_keys: list[str] = field(default_factory=list)
-    created_at: datetime = field(default_factory=datetime.utcnow)
-    updated_at: datetime = field(default_factory=datetime.utcnow)
+    created_at: datetime = field(default_factory=utc_now_naive)
+    updated_at: datetime = field(default_factory=utc_now_naive)
 
     @property
     def has_prd(self) -> bool:
@@ -46,8 +48,8 @@ class Epic:
     summary: str = ""
     plan_content: str = ""
     task_keys: list[str] = field(default_factory=list)
-    created_at: datetime = field(default_factory=datetime.utcnow)
-    updated_at: datetime = field(default_factory=datetime.utcnow)
+    created_at: datetime = field(default_factory=utc_now_naive)
+    updated_at: datetime = field(default_factory=utc_now_naive)
 
     @property
     def has_plan(self) -> bool:
@@ -67,8 +69,8 @@ class Task:
     target_repo: str = ""
     acceptance_criteria: list[str] = field(default_factory=list)
     pr_url: str | None = None
-    created_at: datetime = field(default_factory=datetime.utcnow)
-    updated_at: datetime = field(default_factory=datetime.utcnow)
+    created_at: datetime = field(default_factory=utc_now_naive)
+    updated_at: datetime = field(default_factory=utc_now_naive)
 
     @property
     def has_implementation_details(self) -> bool:
diff --git a/src/forge/queue/models.py b/src/forge/queue/models.py
index 106dbecae..0ad4ba4c4 100644
--- a/src/forge/queue/models.py
+++ b/src/forge/queue/models.py
@@ -23,6 +23,7 @@
     ReviewState,
 )
 from forge.models.events import EventSource
+from forge.utils.time import utc_now_naive
 
 
 @dataclass
@@ -36,7 +37,7 @@ class QueueMessage:
     ticket_key: str
     payload: dict[str, Any] = field(default_factory=dict)
     normalized_event: dict[str, Any] | None = None
-    timestamp: datetime = field(default_factory=datetime.utcnow)
+    timestamp: datetime = field(default_factory=utc_now_naive)
     retry_count: int = 0
 
     def to_dict(self) -> dict[str, str]:
@@ -78,7 +79,7 @@ def from_redis(cls, message_id: str, data: dict[str, str]) -> "QueueMessage":
             ticket_key=data.get("ticket_key", ""),
             payload=json.loads(data.get("payload", "{}")),
             normalized_event=json.loads(normalized_event_raw) if normalized_event_raw else None,
-            timestamp=datetime.fromisoformat(data.get("timestamp", datetime.utcnow().isoformat())),
+            timestamp=datetime.fromisoformat(data.get("timestamp", utc_now_naive().isoformat())),
             retry_count=int(data.get("retry_count", "0")),
         )
 
diff --git a/src/forge/queue/retry.py b/src/forge/queue/retry.py
index eb537809d..cc2e50149 100644
--- a/src/forge/queue/retry.py
+++ b/src/forge/queue/retry.py
@@ -9,6 +9,7 @@
 
 from forge.orchestrator.checkpointer import get_redis_client
 from forge.queue.models import QueueMessage
+from forge.utils.time import utc_now_naive
 
 logger = logging.getLogger(__name__)
 
@@ -57,7 +58,7 @@
 
 
 def _now_timestamp() -> float:
-    return datetime.utcnow().timestamp()
+    return utc_now_naive().timestamp()
 
 
 @dataclass
@@ -150,7 +151,7 @@ async def enqueue_for_retry(
             INITIAL_RETRY_DELAY_SECONDS * (RETRY_BACKOFF_MULTIPLIER ** (attempt - 1)),
             MAX_RETRY_DELAY_SECONDS,
         )
-        next_retry = datetime.utcnow() + timedelta(seconds=delay)
+        next_retry = utc_now_naive() + timedelta(seconds=delay)
 
         entry = RetryEntry(
             message=message,
@@ -191,7 +192,7 @@ async def _move_to_dead_letter(
             "message": msg,
             "error": error,
             "attempts": attempt,
-            "failed_at": datetime.utcnow().isoformat(),
+            "failed_at": utc_now_naive().isoformat(),
         }
 
         serialized = json.dumps(entry)
@@ -382,7 +383,7 @@ async def requeue_dead_letter(self, index: int) -> bool:
             entry = RetryEntry(
                 message=message,
                 attempt=1,
-                next_retry=datetime.utcnow() + timedelta(seconds=INITIAL_RETRY_DELAY_SECONDS),
+                next_retry=utc_now_naive() + timedelta(seconds=INITIAL_RETRY_DELAY_SECONDS),
                 last_error="Requeued from dead-letter",
             )
             await redis.zadd(
diff --git a/src/forge/utils/logging.py b/src/forge/utils/logging.py
index dff57aaf2..b54f1f8f0 100644
--- a/src/forge/utils/logging.py
+++ b/src/forge/utils/logging.py
@@ -3,12 +3,12 @@
 import json
 import logging
 import sys
-from datetime import datetime
 from logging.handlers import RotatingFileHandler
 from pathlib import Path
 from typing import Any
 
 from forge.config import get_settings
+from forge.utils.time import utc_now_naive
 
 
 class StructuredFormatter(logging.Formatter):
@@ -49,7 +49,7 @@ def format(self, record: logging.LogRecord) -> str:
             JSON-formatted log string.
         """
         log_data: dict[str, Any] = {
-            "timestamp": datetime.utcnow().isoformat() + "Z",
+            "timestamp": utc_now_naive().isoformat() + "Z",
             "level": record.levelname,
             "logger": record.name,
             "message": record.getMessage(),
diff --git a/src/forge/utils/time.py b/src/forge/utils/time.py
new file mode 100644
index 000000000..53f7f1823
--- /dev/null
+++ b/src/forge/utils/time.py
@@ -0,0 +1,8 @@
+"""Time helpers for existing timezone-free UTC persistence fields."""
+
+from datetime import UTC, datetime
+
+
+def utc_now_naive() -> datetime:
+    """Return UTC without tzinfo for legacy timestamp wire formats."""
+    return datetime.now(UTC).replace(tzinfo=None)
diff --git a/src/forge/workflow/base.py b/src/forge/workflow/base.py
index 215b4ea42..0e27ecc82 100644
--- a/src/forge/workflow/base.py
+++ b/src/forge/workflow/base.py
@@ -2,13 +2,13 @@
 
 import operator
 from abc import ABC, abstractmethod
-from datetime import datetime
 from typing import Annotated, Any, TypedDict
 
 from langgraph.graph import StateGraph
 from langgraph.graph.message import add_messages
 
 from forge.models.workflow import TicketType
+from forge.utils.time import utc_now_naive
 from forge.workflow.pr_state import PullRequestState
 
 
@@ -254,7 +254,7 @@ def build_graph(self) -> StateGraph[Any]:
 
     def create_initial_state(self, ticket_key: str, **kwargs: Any) -> dict[str, Any]:
         """Create initial state for a new workflow run."""
-        now = datetime.utcnow().isoformat()
+        now = utc_now_naive().isoformat()
         return {
             "thread_id": ticket_key,
             "ticket_key": ticket_key,
diff --git a/src/forge/workflow/bug/state.py b/src/forge/workflow/bug/state.py
index cb5f2d5c5..984e6a617 100644
--- a/src/forge/workflow/bug/state.py
+++ b/src/forge/workflow/bug/state.py
@@ -1,10 +1,10 @@
 """Bug workflow state definition."""
 
-from datetime import datetime
 from typing import Any, cast
 
 from forge.config import get_settings
 from forge.models.workflow import TicketType
+from forge.utils.time import utc_now_naive
 from forge.workflow.base import (
     BaseState,
     CIIntegrationState,
@@ -66,7 +66,7 @@ class BugState(
 
 def create_initial_bug_state(ticket_key: str, **kwargs: Any) -> BugState:
     """Create initial state for a new Bug workflow run."""
-    now = datetime.utcnow().isoformat()
+    now = utc_now_naive().isoformat()
     settings = get_settings()
 
     # Default values - can be overridden by kwargs
diff --git a/src/forge/workflow/feature/state.py b/src/forge/workflow/feature/state.py
index df346a39c..01f8b22d8 100644
--- a/src/forge/workflow/feature/state.py
+++ b/src/forge/workflow/feature/state.py
@@ -1,11 +1,11 @@
 """Feature workflow state definition."""
 
-from datetime import datetime
 from typing import Any, cast
 
 from forge.config import get_settings
 from forge.models.draft import ForgeDecompositionDraft
 from forge.models.workflow import TicketType
+from forge.utils.time import utc_now_naive
 from forge.workflow.base import (
     BaseState,
     CIIntegrationState,
@@ -77,7 +77,7 @@ class FeatureState(
 
 def create_initial_feature_state(ticket_key: str, **kwargs: Any) -> FeatureState:
     """Create initial state for a new Feature workflow run."""
-    now = datetime.utcnow().isoformat()
+    now = utc_now_naive().isoformat()
     settings = get_settings()
 
     # Default values - can be overridden by kwargs
diff --git a/src/forge/workflow/task_takeover/state.py b/src/forge/workflow/task_takeover/state.py
index 9aa7e1435..7508423cc 100644
--- a/src/forge/workflow/task_takeover/state.py
+++ b/src/forge/workflow/task_takeover/state.py
@@ -1,9 +1,9 @@
 """Task Takeover workflow state definition."""
 
-from datetime import datetime
 from typing import Any, TypedDict, cast
 
 from forge.models.workflow import TicketType
+from forge.utils.time import utc_now_naive
 from forge.workflow.base import (
     BaseState,
     CIIntegrationState,
@@ -38,7 +38,7 @@ class TaskTakeoverState(
 
 def create_initial_task_takeover_state(ticket_key: str, **kwargs: Any) -> TaskTakeoverState:
     """Create initial state for a new Task Takeover workflow run."""
-    now = datetime.utcnow().isoformat()
+    now = utc_now_naive().isoformat()
     defaults: dict[str, Any] = {
         "thread_id": ticket_key,
         "ticket_key": ticket_key,
diff --git a/src/forge/workflow/utils/__init__.py b/src/forge/workflow/utils/__init__.py
index 504274fd9..95b706e93 100644
--- a/src/forge/workflow/utils/__init__.py
+++ b/src/forge/workflow/utils/__init__.py
@@ -1,10 +1,10 @@
 """Utility functions for workflow state management and comment classification."""
 
-from datetime import datetime
 from typing import Any
 
 from langgraph.graph import END
 
+from forge.utils.time import utc_now_naive
 from forge.workflow.utils.comment_classifier import (
     CommentType,
     classify_comment,
@@ -61,7 +61,7 @@ def resolve_shared_resume_node(current_node: str) -> str | None:
 
 def update_state_timestamp(state: dict[str, Any]) -> dict[str, Any]:
     """Update the state timestamp."""
-    return {**state, "updated_at": datetime.utcnow().isoformat()}
+    return {**state, "updated_at": utc_now_naive().isoformat()}
 
 
 def set_paused(state: dict[str, Any], node_name: str) -> dict[str, Any]:
@@ -70,7 +70,7 @@ def set_paused(state: dict[str, Any], node_name: str) -> dict[str, Any]:
         **state,
         "current_node": node_name,
         "is_paused": True,
-        "updated_at": datetime.utcnow().isoformat(),
+        "updated_at": utc_now_naive().isoformat(),
     }
 
 
@@ -79,7 +79,7 @@ def resume_state(state: dict[str, Any]) -> dict[str, Any]:
     return {
         **state,
         "is_paused": False,
-        "updated_at": datetime.utcnow().isoformat(),
+        "updated_at": utc_now_naive().isoformat(),
     }
 
 
@@ -89,7 +89,7 @@ def set_error(state: dict[str, Any], error: str) -> dict[str, Any]:
         **state,
         "last_error": error,
         "retry_count": state.get("retry_count", 0) + 1,
-        "updated_at": datetime.utcnow().isoformat(),
+        "updated_at": utc_now_naive().isoformat(),
     }
 
 
diff --git a/tests/conftest.py b/tests/conftest.py
index 1cab8332e..0567dabdf 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -1,6 +1,14 @@
 """Shared test fixtures for Forge test suite."""
 
+import asyncio
 import os
+from collections.abc import AsyncGenerator, Generator
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+import pytest_asyncio
+from httpx import ASGITransport, AsyncClient
 
 # Set dummy environment variables for Pydantic Settings validation during test initialization
 os.environ.setdefault("JIRA_BASE_URL", "https://test.atlassian.net")
@@ -11,18 +19,29 @@
 os.environ.setdefault("LLM_MODEL", "claude-3-5-sonnet-20241022")
 os.environ.setdefault("ANTHROPIC_API_KEY", "test-key")
 
-from collections.abc import AsyncGenerator, Generator
-from pathlib import Path
-from unittest.mock import AsyncMock, MagicMock
-
-import pytest
-import pytest_asyncio
-from httpx import ASGITransport, AsyncClient
-
 from forge.config import Settings
 from forge.integrations.github.client import PullRequestCreationResult
 from forge.main import app
 
+_session_event_loop: asyncio.AbstractEventLoop | None = None
+
+
+def pytest_sessionstart(session: pytest.Session) -> None:
+    """Own the loop pytest-asyncio saves as the pre-fixture policy loop."""
+    global _session_event_loop
+    del session
+    _session_event_loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(_session_event_loop)
+
+
+def pytest_sessionfinish(session: pytest.Session, exitstatus: int) -> None:
+    global _session_event_loop
+    del session, exitstatus
+    if _session_event_loop is not None:
+        _session_event_loop.close()
+        _session_event_loop = None
+
+
 _TEST_DIRECTORY_MARKERS = {
     "unit": "unit",
     "contracts": "contract",
diff --git a/tests/fixtures/workflow_states.py b/tests/fixtures/workflow_states.py
index 1f1a8f4e2..9dcc2b654 100644
--- a/tests/fixtures/workflow_states.py
+++ b/tests/fixtures/workflow_states.py
@@ -1,16 +1,16 @@
 """Pre-built WorkflowState snapshots for testing."""
 
 from copy import deepcopy
-from datetime import datetime
 from typing import Any
 
 from forge.models.workflow import TicketType
+from forge.utils.time import utc_now_naive
 from forge.workflow.feature.state import FeatureState as WorkflowState
 
 
 def _timestamp() -> str:
     """Get current timestamp in ISO format."""
-    return datetime.utcnow().isoformat()
+    return utc_now_naive().isoformat()
 
 
 # Base state template
diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py
index 9c6b00dff..a2e22eff9 100644
--- a/tests/integration/conftest.py
+++ b/tests/integration/conftest.py
@@ -9,7 +9,6 @@
 import redis.asyncio as aioredis
 import respx
 from httpx import ASGITransport, AsyncClient, Response
-from testcontainers.redis import RedisContainer
 
 from forge.config import Settings
 from forge.main import app
@@ -207,6 +206,8 @@ def redis_container():
     if not _container_runtime_available():
         pytest.skip("Podman/Docker not available - skipping testcontainers tests")
 
+    from testcontainers.redis import RedisContainer
+
     with RedisContainer("redis/redis-stack-server:7.4.0-v3") as container:
         yield container
 
diff --git a/tests/integration/redis/test_queue_integration.py b/tests/integration/redis/test_queue_integration.py
index 13b226d4e..626df705d 100644
--- a/tests/integration/redis/test_queue_integration.py
+++ b/tests/integration/redis/test_queue_integration.py
@@ -6,7 +6,7 @@
 
 import asyncio
 import json
-from datetime import datetime, timedelta
+from datetime import UTC, datetime, timedelta
 
 import pytest
 
@@ -281,7 +281,7 @@ async def test_due_entry_is_claimed_by_only_one_worker(self, redis_client):
         entry = RetryEntry(
             message=message,
             attempt=1,
-            next_retry=datetime.utcnow() - timedelta(seconds=1),
+            next_retry=datetime.now(UTC).replace(tzinfo=None) - timedelta(seconds=1),
             last_error="temporary failure",
         )
         serialized_entry = json.dumps(entry.to_dict())
@@ -304,7 +304,7 @@ async def test_due_entry_is_claimed_by_only_one_worker(self, redis_client):
         assert sum(claim[0].message.event_id == message.event_id for claim in claims if claim) == 1
         lease_score = await redis_client.zscore(RETRY_QUEUE_KEY, serialized_entry)
         assert lease_score is not None
-        assert lease_score > datetime.utcnow().timestamp()
+        assert lease_score > datetime.now(UTC).replace(tzinfo=None).timestamp()
 
 
 @pytest.mark.integration
diff --git a/tests/unit/integrations/agents/test_run_task_tracing.py b/tests/unit/integrations/agents/test_run_task_tracing.py
index 71f1d9c8e..451673aab 100644
--- a/tests/unit/integrations/agents/test_run_task_tracing.py
+++ b/tests/unit/integrations/agents/test_run_task_tracing.py
@@ -10,6 +10,7 @@
 import pytest
 
 from forge.integrations.agents.agent import ForgeAgent
+from forge.integrations.agents.structured_outputs import TaskGeneration
 
 
 @pytest.fixture
@@ -34,6 +35,16 @@ def _metrics_patches():
 class TestRunTaskTraceResolution:
     """run_task() resolves trace fields and forwards them to _run_agent()."""
 
+    @pytest.fixture(autouse=True)
+    def fixed_recursion_limit_for_trace_tests(self):
+        with patch(
+            "forge.integrations.agents.agent.resolve_agent_recursion_limit_for_project",
+            new_callable=AsyncMock,
+            create=True,
+        ) as resolver:
+            resolver.return_value = 100
+            yield
+
     @pytest.mark.asyncio
     async def test_builds_trace_state_from_context_and_system_prompt(
         self, agent: ForgeAgent
@@ -233,3 +244,115 @@ async def test_project_policy_is_fetched_for_each_execution(self, agent: ForgeAg
         assert jira.get_project_property.await_count == 4
         assert mock_run.await_args_list[0].kwargs["model_target"].model == "gemini-pro"
         assert mock_run.await_args_list[1].kwargs["model_target"].model == "gemini-flash"
+
+
+class TestRunTaskRecursionResolution:
+    @pytest.mark.asyncio
+    async def test_project_override_is_resolved_once_and_forwarded(self, agent: ForgeAgent):
+        jira = MagicMock()
+        jira.get_project_property = AsyncMock(return_value=75)
+        jira.close = AsyncMock()
+        with (
+            patch.object(agent, "_run_agent", new_callable=AsyncMock) as run,
+            patch("forge.integrations.agents.agent.load_prompt", return_value="prompt"),
+            patch("forge.integrations.jira.client.JiraClient", return_value=jira),
+        ):
+            run.return_value = "ok"
+            await agent.run_task(
+                task="generate-tasks", prompt="test", context={"ticket_key": "PROJ-42"}
+            )
+
+        assert run.await_args.kwargs["recursion_limit"] == 75
+        jira.get_project_property.assert_awaited_once_with("PROJ", "forge.agent_recursion_limit")
+        jira.close.assert_awaited_once()
+
+    @pytest.mark.asyncio
+    async def test_trace_only_ticket_uses_project_limit(self, agent: ForgeAgent):
+        jira = MagicMock()
+        jira.get_project_property = AsyncMock(return_value=80)
+        jira.close = AsyncMock()
+        with (
+            patch.object(agent, "_run_agent", new_callable=AsyncMock) as run,
+            patch("forge.integrations.agents.agent.load_prompt", return_value="prompt"),
+            patch("forge.integrations.jira.client.JiraClient", return_value=jira),
+        ):
+            run.return_value = "ok"
+            await agent.run_task(
+                task="generate-tasks",
+                prompt="test",
+                trace_context={"ticket_key": "PROJ-42"},
+            )
+
+        assert run.await_args.kwargs["recursion_limit"] == 80
+        assert run.await_args.kwargs["ticket_key"] == "PROJ-42"
+
+    @pytest.mark.asyncio
+    async def test_no_ticket_uses_global_without_jira(self, agent: ForgeAgent):
+        agent.settings.agent_recursion_limit = 120
+        with (
+            patch.object(agent, "_run_agent", new_callable=AsyncMock) as run,
+            patch("forge.integrations.agents.agent.load_prompt", return_value="prompt"),
+            patch("forge.integrations.jira.client.JiraClient") as jira_class,
+        ):
+            run.return_value = "ok"
+            await agent.run_task(task="question", prompt="test")
+
+        assert run.await_args.kwargs["recursion_limit"] == 120
+        jira_class.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_invalid_project_limit_fails_before_agent_creation(self, agent: ForgeAgent):
+        jira = MagicMock()
+        jira.get_project_property = AsyncMock(return_value=True)
+        jira.close = AsyncMock()
+        with (
+            patch.object(agent, "_run_agent", new_callable=AsyncMock) as run,
+            patch("forge.integrations.agents.agent.load_prompt", return_value="prompt"),
+            patch("forge.integrations.jira.client.JiraClient", return_value=jira),
+            pytest.raises(ValueError, match="forge.agent_recursion_limit for PROJ"),
+        ):
+            run.return_value = "ok"
+            await agent.run_task(
+                task="generate-tasks", prompt="test", context={"ticket_key": "PROJ-42"}
+            )
+
+        run.assert_not_awaited()
+        jira.close.assert_awaited_once()
+
+    @pytest.mark.asyncio
+    async def test_resolved_limit_stays_fixed_across_tool_fallback(self, agent: ForgeAgent):
+        jira = MagicMock()
+        jira.get_project_property = AsyncMock(return_value=75)
+        jira.close = AsyncMock()
+        native = AsyncMock()
+        native.ainvoke.side_effect = ValueError("native output unavailable")
+        fallback = AsyncMock()
+        fallback.ainvoke.return_value = {
+            "messages": [],
+            "structured_response": {
+                "tasks": [
+                    {
+                        "summary": "Offline",
+                        "description": "No external work",
+                        "repo": "example/repo",
+                    }
+                ]
+            },
+        }
+        with (
+            patch.object(agent, "_create_agent_async", side_effect=[native, fallback]) as create,
+            patch("forge.integrations.agents.agent.load_prompt", return_value="prompt"),
+            patch("forge.integrations.jira.client.JiraClient", return_value=jira),
+        ):
+            result = await agent.run_task(
+                task="generate-tasks",
+                prompt="test",
+                context={"ticket_key": "PROJ-42"},
+                response_schema=TaskGeneration,
+            )
+
+        assert isinstance(result, TaskGeneration)
+        assert jira.get_project_property.await_count == 1
+        assert create.await_count == 2
+        assert native.ainvoke.await_args.kwargs["config"]["recursion_limit"] == 75
+        assert fallback.ainvoke.await_args.kwargs["config"]["recursion_limit"] == 75
diff --git a/tests/unit/integrations/agents/test_structured_output.py b/tests/unit/integrations/agents/test_structured_output.py
index 45357e53d..4c841610a 100644
--- a/tests/unit/integrations/agents/test_structured_output.py
+++ b/tests/unit/integrations/agents/test_structured_output.py
@@ -1,9 +1,17 @@
+from typing import Any
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import anthropic
 import httpx
 import pytest
+from deepagents import create_deep_agent
+from langchain.agents import create_agent
 from langchain.agents.structured_output import ProviderStrategy, ToolStrategy
+from langchain_core.language_models.chat_models import BaseChatModel
+from langchain_core.messages import AIMessage
+from langchain_core.outputs import ChatGeneration, ChatResult
+from langchain_core.tools import tool
+from langgraph.errors import GraphRecursionError
 from pydantic import BaseModel, ConfigDict
 
 from forge.integrations.agents.agent import ForgeAgent
@@ -138,6 +146,41 @@ async def test_vertex_transient_failure_retries_same_native_strategy(error: Exce
     sleep.assert_awaited_once()
 
 
+@tool
+async def repeat_probe() -> str:
+    """Return a harmless value for a synthetic repeating tool loop."""
+    return "again"
+
+
+class RepeatingToolModel(BaseChatModel):
+    calls: int = 0
+
+    @property
+    def _llm_type(self) -> str:
+        return "repeating-tool-test"
+
+    def bind_tools(self, _tools: Any, **_kwargs: Any) -> Any:
+        return self
+
+    def _generate(self, _messages: Any, **_kwargs: Any) -> ChatResult:
+        self.calls += 1
+        return ChatResult(
+            generations=[
+                ChatGeneration(
+                    message=AIMessage(
+                        content="",
+                        tool_calls=[
+                            {"name": "repeat_probe", "args": {}, "id": f"call-{self.calls}"}
+                        ],
+                    )
+                )
+            ]
+        )
+
+    async def _agenerate(self, messages: Any, **kwargs: Any) -> ChatResult:
+        return self._generate(messages, **kwargs)
+
+
 @pytest.mark.asyncio
 async def test_tool_loop_returns_validated_structured_response() -> None:
     forge = ForgeAgent()
@@ -190,3 +233,138 @@ async def test_invalid_fallback_response_raises_actionable_validation_error() ->
         pytest.raises(ValueError, match="reason"),
     ):
         await forge._run_agent("prompt", "system", response_schema=Decision)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("response_schema", [None, Decision])
+async def test_graph_recursion_exhaustion_never_retries_or_falls_back(
+    response_schema: type[Decision] | None,
+) -> None:
+    forge = ForgeAgent()
+    graph = AsyncMock()
+    graph.ainvoke.side_effect = GraphRecursionError("recursion limit reached")
+
+    with (
+        patch.object(forge, "_create_agent_async", return_value=graph) as create,
+        patch("forge.integrations.agents.agent.asyncio.sleep", new_callable=AsyncMock) as sleep,
+        pytest.raises(GraphRecursionError),
+    ):
+        await forge._run_agent(
+            "prompt",
+            "system",
+            response_schema=response_schema,
+            recursion_limit=4,
+            ticket_key="PROJ-42",
+            trace_name="task:generate-tasks",
+        )
+
+    create.assert_awaited_once()
+    graph.ainvoke.assert_awaited_once()
+    sleep.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_explicit_limit_overrides_tracing_config_without_losing_callbacks() -> None:
+    forge = ForgeAgent()
+    graph = AsyncMock()
+    graph.ainvoke.return_value = {"messages": []}
+    with (
+        patch.object(forge, "_create_agent_async", return_value=graph),
+        patch(
+            "forge.integrations.agents.agent.get_langfuse_config",
+            return_value={
+                "recursion_limit": 999,
+                "callbacks": ["trace-callback"],
+                "configurable": {"thread_id": "tracing-overwrite"},
+            },
+        ),
+    ):
+        await forge._run_agent("prompt", "system", recursion_limit=4)
+
+    config = graph.ainvoke.await_args.kwargs["config"]
+    assert config["recursion_limit"] == 4
+    assert config["callbacks"] == ["trace-callback"]
+    assert config["configurable"]["thread_id"] != "tracing-overwrite"
+    assert config["configurable"]["thread_id"]
+
+
+@pytest.mark.asyncio
+async def test_fallback_and_transient_retry_keep_same_limit() -> None:
+    forge = ForgeAgent()
+    native = AsyncMock()
+    native.ainvoke.side_effect = ValueError("native structured output unavailable")
+    fallback = AsyncMock()
+    fallback.ainvoke.return_value = {
+        "messages": [],
+        "structured_response": {"accepted": True, "reason": "valid"},
+    }
+    with patch.object(forge, "_create_agent_async", side_effect=[native, fallback]):
+        await forge._run_agent("prompt", "system", response_schema=Decision, recursion_limit=7)
+
+    assert native.ainvoke.await_args.kwargs["config"]["recursion_limit"] == 7
+    assert fallback.ainvoke.await_args.kwargs["config"]["recursion_limit"] == 7
+
+    retry_graph = AsyncMock()
+    retry_graph.ainvoke.side_effect = [
+        TimeoutError("request timed out"),
+        {"messages": [{"content": "done"}]},
+    ]
+    with (
+        patch.object(forge, "_create_agent_async", return_value=retry_graph),
+        patch("forge.integrations.agents.agent.asyncio.sleep", new_callable=AsyncMock),
+    ):
+        await forge._run_agent("prompt", "system", recursion_limit=7)
+
+    assert retry_graph.ainvoke.await_count == 2
+    assert all(
+        call.kwargs["config"]["recursion_limit"] == 7
+        for call in retry_graph.ainvoke.await_args_list
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("mode", ["unstructured", "native", "fallback"])
+async def test_real_repeating_tool_graph_exhausts_current_invocation_only(mode: str) -> None:
+    forge = ForgeAgent()
+    model = RepeatingToolModel()
+    response_schema = Decision if mode != "unstructured" else None
+    strategy = (
+        ToolStrategy(Decision)
+        if mode == "fallback"
+        else ProviderStrategy(Decision)
+        if mode == "native"
+        else None
+    )
+    graph = create_agent(model, tools=[repeat_probe], response_format=strategy)
+    native = AsyncMock()
+    native.ainvoke.side_effect = ValueError("native format unsupported")
+    created = [native, graph] if mode == "fallback" else [graph]
+
+    with (
+        patch.object(forge, "_create_agent_async", side_effect=created) as factory,
+        patch("forge.integrations.agents.agent.asyncio.sleep", new_callable=AsyncMock) as sleep,
+        pytest.raises(GraphRecursionError),
+    ):
+        await forge._run_agent(
+            "repeat tool forever", "test", response_schema=response_schema, recursion_limit=4
+        )
+
+    assert model.calls > 0
+    assert factory.await_count == len(created)
+    sleep.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_real_deep_agent_graph_honors_invocation_limit_below_default() -> None:
+    forge = ForgeAgent()
+    model = RepeatingToolModel()
+    graph = create_deep_agent(model=model, tools=[repeat_probe], system_prompt="Offline test")
+
+    with (
+        patch.object(forge, "_create_agent_async", return_value=graph) as factory,
+        pytest.raises(GraphRecursionError),
+    ):
+        await forge._run_agent("repeat tool forever", "test", recursion_limit=4)
+
+    assert model.calls > 0
+    factory.assert_awaited_once()
diff --git a/tests/unit/queue/test_consumer.py b/tests/unit/queue/test_consumer.py
index c0ed1fba1..3c4fe07eb 100644
--- a/tests/unit/queue/test_consumer.py
+++ b/tests/unit/queue/test_consumer.py
@@ -5,7 +5,7 @@
 
 import asyncio
 import time
-from datetime import datetime
+from datetime import UTC, datetime
 from unittest.mock import AsyncMock, MagicMock
 
 import pytest
@@ -63,7 +63,7 @@ def _make_message(
         event_type="jira:issue_updated",
         ticket_key=ticket_key,
         payload={},
-        timestamp=datetime.utcnow(),
+        timestamp=datetime.now(UTC).replace(tzinfo=None),
     )
 
 
diff --git a/tests/unit/queue/test_consumer_retry.py b/tests/unit/queue/test_consumer_retry.py
index 991c00927..2143b8858 100644
--- a/tests/unit/queue/test_consumer_retry.py
+++ b/tests/unit/queue/test_consumer_retry.py
@@ -1,6 +1,6 @@
 """Unit tests for QueueConsumer retry integration."""
 
-from datetime import datetime
+from datetime import UTC, datetime
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
@@ -246,7 +246,7 @@ async def test_successful_retry_removes_entry(self):
         entry = RetryEntry(
             message=message,
             attempt=1,
-            next_retry=datetime.utcnow(),
+            next_retry=datetime.now(UTC).replace(tzinfo=None),
             last_error="boom",
         )
 
@@ -278,7 +278,7 @@ async def test_failed_retry_reenqueues(self):
         entry = RetryEntry(
             message=message,
             attempt=1,
-            next_retry=datetime.utcnow(),
+            next_retry=datetime.now(UTC).replace(tzinfo=None),
             last_error="boom",
         )
 
@@ -320,7 +320,7 @@ async def test_exhausted_retry_invokes_terminal_callback(self):
         entry = RetryEntry(
             message=message,
             attempt=3,
-            next_retry=datetime.utcnow(),
+            next_retry=datetime.now(UTC).replace(tzinfo=None),
             last_error="still broken",
         )
         notification = TerminalNotification(
@@ -358,7 +358,7 @@ async def test_terminal_callback_failure_still_acknowledges_retry(self):
         entry = RetryEntry(
             message=message,
             attempt=3,
-            next_retry=datetime.utcnow(),
+            next_retry=datetime.now(UTC).replace(tzinfo=None),
             last_error="still broken",
         )
 
diff --git a/tests/unit/queue/test_retry_queue.py b/tests/unit/queue/test_retry_queue.py
index 962ab626a..870ac6508 100644
--- a/tests/unit/queue/test_retry_queue.py
+++ b/tests/unit/queue/test_retry_queue.py
@@ -2,7 +2,7 @@
 
 import asyncio
 import json
-from datetime import datetime
+from datetime import UTC, datetime
 from unittest.mock import AsyncMock, patch
 
 import pytest
@@ -252,7 +252,7 @@ async def test_concurrent_workers_claim_notification_once(self):
                 "message": {**message.to_dict(), "message_id": message.message_id},
                 "error": "final failure",
                 "attempts": 4,
-                "failed_at": datetime.utcnow().isoformat(),
+                "failed_at": datetime.now(UTC).replace(tzinfo=None).isoformat(),
             }
         ).encode()
         claim_lock = asyncio.Lock()
@@ -342,7 +342,7 @@ async def test_removes_stale_terminal_notification(self):
                 "message": {**message.to_dict(), "message_id": message.message_id},
                 "error": "final failure",
                 "attempts": 4,
-                "failed_at": datetime.utcnow().isoformat(),
+                "failed_at": datetime.now(UTC).replace(tzinfo=None).isoformat(),
             }
         ).encode()
         redis = make_redis_mock()
diff --git a/tests/unit/test_agent_recursion_policy.py b/tests/unit/test_agent_recursion_policy.py
new file mode 100644
index 000000000..f8dba0993
--- /dev/null
+++ b/tests/unit/test_agent_recursion_policy.py
@@ -0,0 +1,261 @@
+"""Host-agent graph limit precedence and bounded Jira property reads."""
+
+import asyncio
+import importlib
+from unittest.mock import AsyncMock, MagicMock
+
+import httpx
+import pytest
+
+
+def _policy():
+    return importlib.import_module("forge.agent_recursion_policy")
+
+
+@pytest.mark.parametrize("raw", [None, 75, 100])
+def test_project_limit_validator_accepts_null_or_positive_integer(raw: int | None) -> None:
+    assert _policy().validate_project_agent_recursion_limit(raw, "AISOS") == raw
+
+
+@pytest.mark.parametrize("raw", [True, False, "75", 1.5, [], {}, 0, -1])
+def test_project_limit_validator_rejects_invalid_non_null_values(raw: object) -> None:
+    with pytest.raises(ValueError, match="forge.agent_recursion_limit for AISOS"):
+        _policy().validate_project_agent_recursion_limit(raw, "AISOS")
+
+
+@pytest.mark.asyncio
+async def test_no_ticket_uses_global_limit_without_jira() -> None:
+    jira = AsyncMock()
+    settings = MagicMock(agent_recursion_limit=150)
+
+    assert (
+        await _policy().resolve_agent_recursion_limit_for_project(settings, None, jira=jira) == 150
+    )
+    jira.get_project_property.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(("raw", "expected"), [(None, 150), (75, 75), (150, 150)])
+async def test_project_override_or_null_global_fallback(
+    raw: int | None, expected: int, caplog: pytest.LogCaptureFixture
+) -> None:
+    jira = AsyncMock()
+    jira.get_project_property.return_value = raw
+    settings = MagicMock(agent_recursion_limit=150)
+
+    result = await _policy().resolve_agent_recursion_limit_for_project(settings, "AISOS", jira=jira)
+
+    assert result == expected
+    jira.get_project_property.assert_awaited_once_with("AISOS", "forge.agent_recursion_limit")
+    assert ("source=global" if raw is None else "source=project") in caplog.text
+
+
+@pytest.mark.asyncio
+async def test_changed_project_property_takes_effect_on_next_run() -> None:
+    jira = AsyncMock()
+    jira.get_project_property.side_effect = [75, 80]
+    settings = MagicMock(agent_recursion_limit=150)
+
+    first = await _policy().resolve_agent_recursion_limit_for_project(settings, "AISOS", jira=jira)
+    second = await _policy().resolve_agent_recursion_limit_for_project(settings, "AISOS", jira=jira)
+
+    assert (first, second) == (75, 80)
+    assert jira.get_project_property.await_count == 2
+
+
+def _http_error(status: int, retry_after: str | None = None) -> httpx.HTTPStatusError:
+    request = httpx.Request("GET", "https://jira.example.test/project/AISOS/properties/limit")
+    response = httpx.Response(
+        status,
+        request=request,
+        headers={"Retry-After": retry_after} if retry_after is not None else {},
+    )
+    return httpx.HTTPStatusError("secret provider detail", request=request, response=response)
+
+
+@pytest.mark.asyncio
+async def test_temporary_jira_read_retries_then_uses_project_override(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    policy = _policy()
+    delays: list[float] = []
+
+    async def record_sleep(delay: float) -> None:
+        delays.append(delay)
+
+    monkeypatch.setattr(policy, "_sleep", record_sleep, raising=False)
+    jira = AsyncMock()
+    jira.get_project_property.side_effect = [_http_error(503), 75]
+
+    result = await policy.read_agent_recursion_property(jira, "AISOS")
+
+    assert result == 75
+    assert jira.get_project_property.await_count == 2
+    assert delays == [0.5]
+
+
+@pytest.mark.asyncio
+async def test_jira_read_stops_after_three_transient_attempts(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    policy = _policy()
+    delays: list[float] = []
+
+    async def record_sleep(delay: float) -> None:
+        delays.append(delay)
+
+    monkeypatch.setattr(policy, "_sleep", record_sleep, raising=False)
+    jira = AsyncMock()
+    jira.get_project_property.side_effect = _http_error(503)
+
+    with pytest.raises(RuntimeError, match="forge.agent_recursion_limit.*AISOS.*3 attempts") as exc:
+        await policy.read_agent_recursion_property(jira, "AISOS")
+
+    assert "secret provider detail" not in str(exc.value)
+    assert jira.get_project_property.await_count == 3
+    assert delays == [0.5, 1.0]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("status", [400, 401, 403])
+async def test_jira_auth_or_request_error_fails_without_retry(status: int) -> None:
+    jira = AsyncMock()
+    jira.get_project_property.side_effect = _http_error(status)
+
+    with pytest.raises(RuntimeError, match=f"forge.agent_recursion_limit.*AISOS.*{status}"):
+        await _policy().read_agent_recursion_property(jira, "AISOS")
+
+    jira.get_project_property.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_jira_404_is_successful_unset() -> None:
+    jira = AsyncMock()
+    jira.get_project_property.side_effect = _http_error(404)
+
+    assert await _policy().read_agent_recursion_property(jira, "AISOS") is None
+    jira.get_project_property.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_jira_retry_after_inside_budget_is_honored(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    policy = _policy()
+    delays: list[float] = []
+
+    async def record_sleep(delay: float) -> None:
+        delays.append(delay)
+
+    monkeypatch.setattr(policy, "_sleep", record_sleep, raising=False)
+    jira = AsyncMock()
+    jira.get_project_property.side_effect = [_http_error(429, "2"), None]
+
+    assert await policy.read_agent_recursion_property(jira, "AISOS") is None
+    assert delays == [2.0]
+
+
+@pytest.mark.asyncio
+async def test_jira_retry_after_beyond_total_budget_fails_without_sleep(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    policy = _policy()
+    sleep = AsyncMock()
+    monkeypatch.setattr(policy, "_sleep", sleep, raising=False)
+    jira = AsyncMock()
+    jira.get_project_property.side_effect = _http_error(429, "31")
+
+    with pytest.raises(RuntimeError, match="forge.agent_recursion_limit.*AISOS.*budget"):
+        await policy.read_agent_recursion_property(jira, "AISOS")
+
+    jira.get_project_property.assert_awaited_once()
+    sleep.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_jira_read_enforces_per_attempt_timeout(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    policy = _policy()
+    timeouts: list[float] = []
+
+    async def timed_attempt(awaitable: object, timeout: float) -> object:
+        timeouts.append(timeout)
+        awaitable.close()
+        raise TimeoutError
+
+    monkeypatch.setattr(policy, "_await_with_timeout", timed_attempt, raising=False)
+    monkeypatch.setattr(policy, "_sleep", AsyncMock(), raising=False)
+    jira = AsyncMock()
+
+    with pytest.raises(RuntimeError, match="3 attempts"):
+        await policy.read_agent_recursion_property(jira, "AISOS")
+
+    assert len(timeouts) == 3
+    assert all(0 < timeout <= 10 for timeout in timeouts)
+
+
+@pytest.mark.asyncio
+async def test_jira_read_stops_at_total_elapsed_budget(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    policy = _policy()
+    clock = {"now": 0.0}
+    monkeypatch.setattr(policy, "_monotonic", lambda: clock["now"], raising=False)
+    jira = AsyncMock()
+
+    async def exceed_budget(*_args: object) -> None:
+        clock["now"] = 31.0
+        raise httpx.ConnectError("secret transport detail")
+
+    jira.get_project_property.side_effect = exceed_budget
+    with pytest.raises(RuntimeError, match="forge.agent_recursion_limit.*AISOS.*budget"):
+        await policy.read_agent_recursion_property(jira, "AISOS")
+
+    jira.get_project_property.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_jira_cancellation_and_malformed_response_do_not_retry() -> None:
+    for error in (asyncio.CancelledError(), ValueError("malformed property JSON")):
+        jira = AsyncMock()
+        jira.get_project_property.side_effect = error
+        with pytest.raises(type(error)):
+            await _policy().read_agent_recursion_property(jira, "AISOS")
+        jira.get_project_property.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_owned_jira_client_closes_on_success_and_failure(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    policy = _policy()
+    settings = MagicMock(agent_recursion_limit=100)
+    clients: list[AsyncMock] = []
+
+    def make_client(**_kwargs: object) -> AsyncMock:
+        client = AsyncMock()
+        client.get_project_property.side_effect = [75] if not clients else [ValueError("bad")]
+        clients.append(client)
+        return client
+
+    monkeypatch.setattr("forge.integrations.jira.client.JiraClient", make_client)
+    assert await policy.resolve_agent_recursion_limit_for_project(settings, "AISOS") == 75
+    with pytest.raises(ValueError):
+        await policy.resolve_agent_recursion_limit_for_project(settings, "AISOS")
+    assert len(clients) == 2
+    for client in clients:
+        client.close.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_injected_jira_client_is_never_closed() -> None:
+    settings = MagicMock(agent_recursion_limit=100)
+    jira = AsyncMock()
+    jira.get_project_property.return_value = 75
+
+    assert (
+        await _policy().resolve_agent_recursion_limit_for_project(settings, "AISOS", jira=jira)
+        == 75
+    )
+    jira.close.assert_not_awaited()
diff --git a/tests/unit/test_cli_config.py b/tests/unit/test_cli_config.py
index 6d6a03864..ebf4ebfd5 100644
--- a/tests/unit/test_cli_config.py
+++ b/tests/unit/test_cli_config.py
@@ -13,7 +13,12 @@
 class TestCLIConfigParserAndRouting:
     """Parser and Command Routing Integration Tests."""
 
-    @patch("forge.cli.cmd_get_config", new_callable=AsyncMock)
+    @pytest.fixture(autouse=True)
+    def _without_event_loop(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        # These tests inspect argument routing; no async handler needs to run.
+        monkeypatch.setattr("forge.cli.asyncio.run", lambda result: result)
+
+    @patch("forge.cli.cmd_get_config", new_callable=MagicMock)
     @patch("forge.cli.setup_logging")
     def test_routing_get_config(self, _mock_setup_logging, mock_cmd):
         """Calling main(['get-config', 'aisos']) routes to cmd_get_config."""
@@ -25,7 +30,7 @@ def test_routing_get_config(self, _mock_setup_logging, mock_cmd):
         assert args.command == "get-config"
         assert args.project_key == "AISOS"  # converts lowercase to uppercase
 
-    @patch("forge.cli.cmd_get_config", new_callable=AsyncMock)
+    @patch("forge.cli.cmd_get_config", new_callable=MagicMock)
     @patch("forge.cli.setup_logging")
     def test_routing_project_config_alias(self, _mock_setup_logging, mock_cmd):
         """Calling main(['project-config', 'aisos']) successfully maps to cmd_get_config."""
@@ -43,7 +48,7 @@ def test_mutual_exclusivity(self, _mock_setup_logging):
         with pytest.raises(SystemExit):
             main(["get-config", "aisos", "--json", "--property", "forge.repos"])
 
-    @patch("forge.cli.cmd_project_setup", new_callable=AsyncMock)
+    @patch("forge.cli.cmd_project_setup", new_callable=MagicMock)
     @patch("forge.cli.setup_logging")
     def test_project_setup_model_all_parsing(self, _mock_setup_logging, mock_cmd):
         mock_cmd.return_value = 0
@@ -55,7 +60,7 @@ def test_project_setup_model_all_parsing(self, _mock_setup_logging, mock_cmd):
         assert args.project_key == "aisos"
         assert args.model_all == "vertex-prod:gemini-pro"
 
-    @patch("forge.cli.cmd_project_setup", new_callable=AsyncMock)
+    @patch("forge.cli.cmd_project_setup", new_callable=MagicMock)
     @patch("forge.cli.setup_logging")
     def test_project_setup_model_removal_parsing(self, _mock_setup_logging, mock_cmd):
         mock_cmd.return_value = 0
@@ -66,7 +71,7 @@ def test_project_setup_model_removal_parsing(self, _mock_setup_logging, mock_cmd
         args = mock_cmd.call_args.args[0]
         assert args.remove_model == ["generate_prd"]
 
-    @patch("forge.cli.cmd_project_setup", new_callable=AsyncMock)
+    @patch("forge.cli.cmd_project_setup", new_callable=MagicMock)
     @patch("forge.cli.setup_logging")
     def test_project_setup_incremental_flags(self, _mock_setup_logging, mock_cmd):
         mock_cmd.return_value = 0
@@ -95,7 +100,7 @@ def test_project_setup_incremental_flags(self, _mock_setup_logging, mock_cmd):
         assert args.remove_prd_proposals_path is True
         assert args.remove_skills is True
 
-    @patch("forge.cli.cmd_project_setup", new_callable=AsyncMock)
+    @patch("forge.cli.cmd_project_setup", new_callable=MagicMock)
     @patch("forge.cli.setup_logging")
     def test_project_setup_json_parsing(self, _mock_setup_logging, mock_cmd):
         mock_cmd.return_value = 0
@@ -109,6 +114,355 @@ def test_project_setup_json_parsing(self, _mock_setup_logging, mock_cmd):
         assert args.json is True
 
 
+class TestAgentRecursionCLI:
+    @staticmethod
+    def setup_args(**overrides):
+        values = TestCLIConfigExecution.setup_args(**overrides)
+        values.project_key = "AISOS"
+        return values
+
+    @staticmethod
+    def mock_config_jira(raw=None, *, error=None, properties=None):
+        jira = MagicMock()
+        properties = properties or {}
+        jira.list_project_properties = AsyncMock(return_value=list(properties))
+
+        async def read(_project, key):
+            if key == "forge.agent_recursion_limit":
+                if error is not None:
+                    raise error
+                return raw
+            return properties.get(key)
+
+        jira.get_project_property = AsyncMock(side_effect=read)
+        jira.close = AsyncMock()
+        return jira
+
+    @pytest.mark.parametrize("recursion_state", ["valid", "malformed", "read_failure"])
+    @pytest.mark.parametrize("property_key", ["forge.repos", "FORGE.REPOS", "forge.custom", None])
+    @pytest.mark.asyncio
+    async def test_unrelated_queries_skip_recursion_property(
+        self, recursion_state, property_key, capsys
+    ):
+        error = None
+        if recursion_state == "read_failure":
+            request = httpx.Request("GET", "https://jira.example.test/property")
+            error = httpx.HTTPStatusError(
+                "secret provider detail",
+                request=request,
+                response=httpx.Response(403, request=request),
+            )
+        jira = self.mock_config_jira(
+            "75" if recursion_state == "malformed" else 75,
+            error=error,
+            properties={
+                "forge.repos": ["example/repo"],
+                "forge.custom": "custom-value",
+                "forge.agent_recursion_limit": None,
+                "forge.model_policy": {
+                    "generate_tasks": {"connection": "default", "model": "claude-opus-4-8"}
+                },
+            },
+        )
+        with patch("forge.integrations.jira.client.JiraClient", return_value=jira):
+            code = await cmd_get_config(
+                SimpleNamespace(
+                    project_key="AISOS",
+                    json=False,
+                    property=property_key,
+                    models=property_key is None,
+                )
+            )
+
+        assert code == 0
+        captured = capsys.readouterr()
+        assert not captured.err
+        if property_key is None:
+            target = json.loads(captured.out)["generate_tasks"]
+            assert target["connection"] == "default"
+            assert target["model"] == "claude-opus-4-8"
+            assert target["policy_source"] == "project"
+        elif property_key.lower() == "forge.repos":
+            assert json.loads(captured.out) == ["example/repo"]
+        else:
+            assert captured.out.strip() == "custom-value"
+        assert all(
+            call.args[1] != "forge.agent_recursion_limit"
+            for call in jira.get_project_property.await_args_list
+        )
+
+    @pytest.mark.parametrize("recursion_state", ["malformed", "read_failure"])
+    @pytest.mark.parametrize("output_mode", ["text", "json", "property", "property_uppercase"])
+    @pytest.mark.asyncio
+    async def test_recursion_queries_remain_fail_closed(self, recursion_state, output_mode, capsys):
+        error = None
+        if recursion_state == "read_failure":
+            request = httpx.Request("GET", "https://jira.example.test/property")
+            error = httpx.HTTPStatusError(
+                "secret provider detail",
+                request=request,
+                response=httpx.Response(403, request=request),
+            )
+        jira = self.mock_config_jira("75", error=error)
+        property_key = {
+            "property": "forge.agent_recursion_limit",
+            "property_uppercase": "FORGE.AGENT_RECURSION_LIMIT",
+        }.get(output_mode)
+        with patch("forge.integrations.jira.client.JiraClient", return_value=jira):
+            code = await cmd_get_config(
+                SimpleNamespace(
+                    project_key="AISOS",
+                    json=output_mode == "json",
+                    property=property_key,
+                    models=False,
+                )
+            )
+
+        assert code == 1
+        captured = capsys.readouterr()
+        assert "forge.agent_recursion_limit" in captured.err
+        assert "secret provider detail" not in captured.out + captured.err
+        if output_mode == "json":
+            entry = json.loads(captured.out)["effective"]["forge.agent_recursion_limit"]
+            assert entry["value"] is None
+            assert entry["source"] == "error"
+        elif output_mode == "text":
+            assert "(error)" in captured.out
+        else:
+            assert not captured.out
+        jira.get_project_property.assert_any_await("AISOS", "forge.agent_recursion_limit")
+
+    @pytest.fixture(autouse=True)
+    def config_settings(self):
+        from forge.config import Settings
+
+        settings = Settings(
+            _env_file=None,
+            jira_base_url="https://jira.example.test",
+            jira_api_token="dummy",
+            jira_user_email="tester@example.test",
+            github_token="dummy",
+            llm_backend="vertex-ai",
+            llm_model="claude-opus-4-8",
+            google_cloud_project="offline-project",
+            agent_recursion_limit=100,
+        )
+        with patch("forge.config.get_settings", return_value=settings):
+            yield settings
+
+    @pytest.mark.asyncio
+    async def test_set_limit_writes_json_integer_and_reports_mutation(self, capsys):
+        jira = MagicMock()
+        jira.set_project_property = AsyncMock()
+        jira.close = AsyncMock()
+        with patch("forge.integrations.jira.client.JiraClient", return_value=jira):
+            code = await cmd_project_setup(self.setup_args(agent_recursion_limit=150, json=True))
+
+        assert code == 0
+        jira.set_project_property.assert_awaited_once_with(
+            "AISOS", "forge.agent_recursion_limit", 150
+        )
+        assert json.loads(capsys.readouterr().out)["mutations"] == {
+            "forge.agent_recursion_limit": {"operation": "set", "value": 150}
+        }
+
+    @pytest.mark.asyncio
+    async def test_clear_absent_limit_succeeds_without_other_mutations(self):
+        jira = MagicMock()
+        jira.delete_project_property = AsyncMock()
+        jira.close = AsyncMock()
+        with patch("forge.integrations.jira.client.JiraClient", return_value=jira):
+            code = await cmd_project_setup(self.setup_args(clear_agent_recursion_limit=True))
+
+        assert code == 0
+        jira.delete_project_property.assert_awaited_once_with(
+            "AISOS", "forge.agent_recursion_limit"
+        )
+        jira.set_project_property.assert_not_called()
+
+    @pytest.mark.parametrize("invalid", [0, -1])
+    @pytest.mark.asyncio
+    async def test_invalid_limit_with_repo_flag_makes_no_jira_calls(self, invalid, capsys):
+        with patch("forge.integrations.jira.client.JiraClient") as jira_class:
+            code = await cmd_project_setup(
+                self.setup_args(agent_recursion_limit=invalid, repo=["org/repo"])
+            )
+
+        assert code == 1
+        jira_class.assert_not_called()
+        assert "positive integer" in capsys.readouterr().err
+
+    @pytest.mark.asyncio
+    async def test_conflicting_limit_flags_make_no_jira_calls(self, capsys):
+        with patch("forge.integrations.jira.client.JiraClient") as jira_class:
+            code = await cmd_project_setup(
+                self.setup_args(agent_recursion_limit=150, clear_agent_recursion_limit=True)
+            )
+
+        assert code == 1
+        jira_class.assert_not_called()
+        assert "cannot be combined" in capsys.readouterr().err
+
+    @pytest.mark.parametrize(
+        ("raw", "expected_source", "expected_value"),
+        [(None, "global", 100), (75, "project", 75), (100, "project", 100)],
+    )
+    @pytest.mark.asyncio
+    async def test_get_config_json_shows_raw_global_and_effective(
+        self, raw, expected_source, expected_value, capsys
+    ):
+        jira = self.mock_config_jira(raw)
+        with patch("forge.integrations.jira.client.JiraClient", return_value=jira):
+            code = await cmd_get_config(
+                SimpleNamespace(project_key="AISOS", json=True, property=None, models=False)
+            )
+
+        assert code == 0
+        data = json.loads(capsys.readouterr().out)
+        assert data["project_properties"]["forge.agent_recursion_limit"] == raw
+        assert data["global_fallbacks"]["AGENT_RECURSION_LIMIT"] == 100
+        assert data["effective"]["forge.agent_recursion_limit"] == {
+            "value": expected_value,
+            "source": expected_source,
+        }
+        assert (
+            sum(
+                call.args[1] == "forge.agent_recursion_limit"
+                for call in jira.get_project_property.await_args_list
+            )
+            == 1
+        )
+
+    @pytest.mark.asyncio
+    async def test_get_config_text_and_property_filter_show_project_limit(self, capsys):
+        jira = self.mock_config_jira(75)
+        with patch("forge.integrations.jira.client.JiraClient", return_value=jira):
+            code = await cmd_get_config(
+                SimpleNamespace(project_key="AISOS", json=False, property=None, models=False)
+            )
+        assert code == 0
+        output = capsys.readouterr().out
+        assert "AGENT_RECURSION_LIMIT:" in output
+        assert "forge.agent_recursion_limit:" in output
+        assert "75 [project]" in output
+
+        jira = self.mock_config_jira(75)
+        with patch("forge.integrations.jira.client.JiraClient", return_value=jira):
+            code = await cmd_get_config(
+                SimpleNamespace(
+                    project_key="AISOS",
+                    json=False,
+                    property="forge.agent_recursion_limit",
+                    models=False,
+                )
+            )
+        assert code == 0
+        assert capsys.readouterr().out.strip() == "75"
+
+    @pytest.mark.parametrize("raw", ["75", True, 0, {"bad": 1}])
+    @pytest.mark.asyncio
+    async def test_malformed_project_limit_is_diagnostic_error(self, raw, capsys):
+        jira = self.mock_config_jira(raw)
+        with patch("forge.integrations.jira.client.JiraClient", return_value=jira):
+            code = await cmd_get_config(
+                SimpleNamespace(project_key="AISOS", json=True, property=None, models=False)
+            )
+
+        assert code == 1
+        captured = capsys.readouterr()
+        data = json.loads(captured.out)
+        assert data["project_properties"]["forge.agent_recursion_limit"] == raw
+        effective = data["effective"]["forge.agent_recursion_limit"]
+        assert effective["value"] is None and effective["source"] == "error"
+        assert "positive integer" in effective["error"]
+        assert "positive integer" in captured.err
+
+    @pytest.mark.asyncio
+    async def test_failed_limit_read_does_not_claim_global_fallback(self, capsys):
+        request = httpx.Request("GET", "https://jira.example.test/property")
+        response = httpx.Response(403, request=request)
+        error = httpx.HTTPStatusError("secret provider detail", request=request, response=response)
+        jira = self.mock_config_jira(error=error)
+        with patch("forge.integrations.jira.client.JiraClient", return_value=jira):
+            code = await cmd_get_config(
+                SimpleNamespace(project_key="AISOS", json=True, property=None, models=False)
+            )
+
+        assert code == 1
+        captured = capsys.readouterr()
+        data = json.loads(captured.out)
+        effective = data["effective"]["forge.agent_recursion_limit"]
+        assert effective["value"] is None and effective["source"] == "error"
+        assert "403" in effective["error"]
+        assert "secret provider detail" not in captured.out + captured.err
+
+    @pytest.mark.asyncio
+    async def test_property_filter_reports_invalid_raw_value_without_global_claim(self, capsys):
+        jira = self.mock_config_jira("75")
+        with patch("forge.integrations.jira.client.JiraClient", return_value=jira):
+            code = await cmd_get_config(
+                SimpleNamespace(
+                    project_key="AISOS",
+                    json=False,
+                    property="forge.agent_recursion_limit",
+                    models=False,
+                )
+            )
+
+        assert code == 1
+        captured = capsys.readouterr()
+        assert not captured.out
+        assert 'Raw forge.agent_recursion_limit: "75"' in captured.err
+        assert "positive integer" in captured.err
+        assert "source=global" not in captured.err
+
+    def test_project_config_alias_routes_recursion_property(self):
+        parsed = []
+
+        def record(args):
+            parsed.append(args)
+            return 0
+
+        with (
+            patch("forge.cli.cmd_get_config", new=record),
+            patch("forge.cli.asyncio.run", side_effect=lambda result: result),
+            patch("forge.cli.setup_logging"),
+        ):
+            assert (
+                main(["project-config", "aisos", "--property", "forge.agent_recursion_limit"]) == 0
+            )
+
+        assert parsed[0].command == "get-config"
+        assert parsed[0].property == "forge.agent_recursion_limit"
+
+    def test_parser_accepts_set_clear_and_rejects_both(self):
+        parsed = []
+
+        def record(args):
+            parsed.append(args)
+            return 0
+
+        with (
+            patch("forge.cli.cmd_project_setup", new=record),
+            patch("forge.cli.asyncio.run", side_effect=lambda result: result),
+            patch("forge.cli.setup_logging"),
+        ):
+            assert main(["project-setup", "aisos", "--agent-recursion-limit", "150"]) == 0
+            assert parsed[-1].agent_recursion_limit == 150
+            assert main(["project-setup", "aisos", "--clear-agent-recursion-limit"]) == 0
+            assert parsed[-1].clear_agent_recursion_limit is True
+            with pytest.raises(SystemExit):
+                main(
+                    [
+                        "project-setup",
+                        "aisos",
+                        "--agent-recursion-limit",
+                        "150",
+                        "--clear-agent-recursion-limit",
+                    ]
+                )
+
+
 class TestCLIConfigExecution:
     """Fallback Semantics, Output Serialization, and Discovery."""
 
@@ -737,7 +1091,9 @@ async def test_malformed_properties_payload_graceful_degradation(self, mock_sett
             client_inst = MagicMock()
             client_inst.list_project_properties = AsyncMock(return_value=["forge.repos"])
             # Return string value "not-a-list" instead of list
-            client_inst.get_project_property = AsyncMock(return_value="not-a-list")
+            client_inst.get_project_property = AsyncMock(
+                side_effect=lambda _pk, key: "not-a-list" if key == "forge.repos" else None
+            )
             client_inst.close = AsyncMock()
             mock.return_value = client_inst
 
@@ -890,28 +1246,29 @@ class Args:
             assert "https://example.com/ref2 - Desc 2" in out
             assert "https://example.com/ref1" not in out
 
-    @patch("forge.cli.cmd_project_setup", new_callable=AsyncMock)
+    @patch("forge.cli.cmd_project_setup", new_callable=MagicMock)
     @patch("forge.cli.setup_logging")
     def test_cli_parser_registers_references(self, _mock_setup_logging, mock_cmd):
         """Verify argparse parser registers --add-reference, --ref-description, --description, --remove-reference, and --list-references."""
         mock_cmd.return_value = 0
-        code = main(
-            [
-                "project-setup",
-                "myproj",
-                "--add-reference",
-                "https://example.com/ref1",
-                "--ref-description",
-                "Desc 1",
-                "--add-reference",
-                "https://example.com/ref2",
-                "--description",
-                "Desc 2",
-                "--remove-reference",
-                "https://example.com/ref3",
-                "--list-references",
-            ]
-        )
+        with patch("forge.cli.asyncio.run", side_effect=lambda result: result):
+            code = main(
+                [
+                    "project-setup",
+                    "myproj",
+                    "--add-reference",
+                    "https://example.com/ref1",
+                    "--ref-description",
+                    "Desc 1",
+                    "--add-reference",
+                    "https://example.com/ref2",
+                    "--description",
+                    "Desc 2",
+                    "--remove-reference",
+                    "https://example.com/ref3",
+                    "--list-references",
+                ]
+            )
         assert code == 0
         mock_cmd.assert_called_once()
         args = mock_cmd.call_args[0][0]
diff --git a/tests/unit/test_config_agent.py b/tests/unit/test_config_agent.py
new file mode 100644
index 000000000..cd99c2679
--- /dev/null
+++ b/tests/unit/test_config_agent.py
@@ -0,0 +1,36 @@
+"""Global host-agent recursion configuration."""
+
+import pytest
+from pydantic import ValidationError
+
+from forge.config import Settings
+
+
+def _settings(**overrides: object) -> Settings:
+    return Settings(
+        _env_file=None,
+        jira_base_url="https://jira.example.test",
+        jira_api_token="dummy",
+        jira_user_email="tester@example.test",
+        github_token="dummy",
+        llm_backend="vertex-ai",
+        llm_model="claude-opus-4-8",
+        google_cloud_project="offline-project",
+        **overrides,
+    )
+
+
+def test_agent_recursion_limit_defaults_to_one_hundred(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delenv("AGENT_RECURSION_LIMIT", raising=False)
+    assert _settings().agent_recursion_limit == 100
+
+
+def test_agent_recursion_limit_reads_environment(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("AGENT_RECURSION_LIMIT", "150")
+    assert _settings().agent_recursion_limit == 150
+
+
+@pytest.mark.parametrize("invalid", [0, -1])
+def test_agent_recursion_limit_rejects_nonpositive_values(invalid: int) -> None:
+    with pytest.raises(ValidationError, match="agent_recursion_limit"):
+        _settings(agent_recursion_limit=invalid)
diff --git a/tests/unit/workflow/stations/test_task_routing.py b/tests/unit/workflow/stations/test_task_routing.py
index c358fac5f..aa908e8fc 100644
--- a/tests/unit/workflow/stations/test_task_routing.py
+++ b/tests/unit/workflow/stations/test_task_routing.py
@@ -13,7 +13,7 @@
     reduce_repository_aggregation,
     reduce_task_routing,
 )
-from forge.workflow.stations.runner import run_serialized
+from forge.workflow.stations.runner import run_serialized_async
 from forge.workflow.stations.task_routing import (
     TaskRoutingOutput,
     run_repository_aggregation_station,
@@ -47,7 +47,7 @@ def test_station_has_no_graph_or_provider_state() -> None:
         first_repository="acme/api",
         task_count=2,
     )
-    assert "current_node" not in outcome.output.model_fields
+    assert "current_node" not in type(outcome.output).model_fields
 
 
 def test_reducer_owns_legacy_topology_mapping() -> None:
@@ -113,10 +113,11 @@ def test_stale_outcome_is_rejected() -> None:
         reduce_task_routing(state, request, outcome)
 
 
-def test_station_runs_from_serialized_fixture_without_control_plane() -> None:
+@pytest.mark.asyncio
+async def test_station_runs_from_serialized_fixture_without_control_plane() -> None:
     request = project_task_routing(_state())
 
-    raw_outcome = run_serialized("task-routing", request.model_dump_json())
+    raw_outcome = await run_serialized_async("task-routing", request.model_dump_json())
 
     assert json.loads(raw_outcome)["output"]["first_repository"] == "acme/api"