feat(code): adopt coding-agent-v1 trace metadata (#4367)

**Why this is needed**
LangSmith groups and filters coding-agent traces by a standard set of
metadata keys — the agent, its version, the conversation turn, and the
repo it ran in. Deep Agents Code didn't emit them, so its traces were
the odd one out: you
couldn't line them up against the other coding agents in dashboards or
evals.

**What this does**
Deep Agents Code now stamps the shared coding-agent-v1 metadata contract
on its LangSmith traces, using the same stable keys as the other
integrations. Each trace now reliably records what agent and version
produced it, which user turn it belongs to, and the repo/git/cwd it ran
in.

**Changes (all in `libs/code`)**
- New `_git.py`: resolves the commit SHA, remote URL, and parsed repo
metadata (provider/name).
- `config.py`: a single `build_coding_agent_metadata()` helper wired
into `build_stream_config`; mirrors `thread_id` to top-level metadata
and keeps existing keys.
- Turn markers (`turn_id` + 1-based `turn_number`) tracked on session
state and passed through the interactive and non-interactive paths.
- Tests for the git helpers, turn markers, and a contract test
validating the metadata block against the vendored coding-agent-v1
validator.

**Known limitation**
On `chat_model` / `create_agent` runs, LangChain core overwrites
`ls_integration` with its own value, so per-run `ls_integration` isn't
`deepagents-code` there. Traces stay identifiable via
`ls_agent_kind="coding_agent"` and root-level `ls_integration`.

**How I verified**
Verified on a [live
trace](https://smith.langchain.com/o/ebbaf2eb-769b-4505-aca2-d11de10372a4/projects/p/32d59ccd-86a1-447c-8a6d-53ee4f75ce62?timeModel=%7B%22duration%22%3A%2230d%22%7D&peek=20260623T212320Z019ef65d-bf93-7450-9690-e0e4c72e09d0&peekedConversationId=019ef65c-5012-7521-a536-725278301be0&trace_id=019ef65d-bf93-7450-9690-e0e4c72e09d0&run_id=20260623T212320Z019ef65d-bf93-7450-9690-e0e4c72e09d0&peeked_trace=20260623T212320288794Z019ef65d-bf93-7450-9690-e0e4c72e09d0&conversationTab=trace&scroll_to=feedback)
— all required keys present per run type.

---

Fixes LSEN-307

---------

Co-authored-by: Mason Daugherty <github@mdrxy.com>
Co-authored-by: Mason Daugherty <mason@langchain.dev>

libs/code/deepagents_code/_git.py

@@ -3,7 +3,10 @@
 from __future__ import annotations
 
 import logging
+import re
 from pathlib import Path
+from typing import NamedTuple
+from urllib.parse import urlparse
 
 logger = logging.getLogger(__name__)
 
@@ -234,3 +237,285 @@ def resolve_git_branch(path: str | Path) -> str:
     if branch is not None:
         return branch
     return read_git_branch_via_subprocess(path)
+
+
+_GIT_SHA_RE = re.compile(r"\A[0-9a-f]{40}(?:[0-9a-f]{24})?\Z")
+"""Matches a full 40-char SHA-1 (or 64-char SHA-256) git object id."""
+
+
+def read_git_commit_sha_from_filesystem(path: str | Path) -> str | None:
+    """Read the current `HEAD` commit SHA from repository metadata.
+
+    Resolves a symbolic `HEAD` (`ref: refs/heads/<branch>`) by reading the
+    loose ref file, then `packed-refs`. A detached `HEAD` already contains the
+    raw SHA.
+
+    Args:
+        path: Directory or file path inside a repository.
+
+    Returns:
+        The full commit SHA (40-char SHA-1 or 64-char SHA-256), an empty string
+            when `path` is not inside a git repository, or `None` when metadata
+            exists but cannot be resolved.
+    """
+    git_dir = find_git_dir(path)
+    if git_dir is None:
+        return ""
+
+    head_path = git_dir / "HEAD"
+    try:
+        head = head_path.read_text(encoding="utf-8").strip()
+    except FileNotFoundError:
+        logger.debug("Git HEAD file not found in %s", git_dir)
+        return None
+    except OSError:
+        logger.debug("Failed to read git HEAD from %s", git_dir, exc_info=True)
+        return None
+
+    if not head:
+        return None
+    if not head.startswith(_GIT_HEAD_REF_PREFIX):
+        # Detached HEAD: the file holds the commit SHA directly.
+        return head if _GIT_SHA_RE.match(head) else None
+
+    ref = head.removeprefix(_GIT_HEAD_REF_PREFIX).strip()
+    if not ref:
+        return None
+
+    loose_ref = git_dir / Path(ref)
+    try:
+        sha = loose_ref.read_text(encoding="utf-8").strip()
+        if _GIT_SHA_RE.match(sha):
+            return sha
+    except FileNotFoundError:
+        pass  # ref may be packed
+    except OSError:
+        logger.debug("Failed to read loose ref %s", loose_ref, exc_info=True)
+        return None
+
+    return _read_packed_ref(git_dir, ref)
+
+
+def _read_packed_ref(git_dir: Path, ref: str) -> str | None:
+    """Resolve a ref to its SHA from `packed-refs`.
+
+    Args:
+        git_dir: Resolved git metadata directory.
+        ref: Full ref name (e.g. `refs/heads/main`).
+
+    Returns:
+        The full commit SHA, or `None` when the ref is absent or unreadable.
+    """
+    try:
+        packed = (git_dir / "packed-refs").read_text(encoding="utf-8")
+    except FileNotFoundError:
+        return None
+    except OSError:
+        logger.debug("Failed to read packed-refs in %s", git_dir, exc_info=True)
+        return None
+
+    for line in packed.splitlines():
+        if not line or line.startswith(("#", "^")):
+            continue
+        sha, _, name = line.partition(" ")
+        if name.strip() == ref and _GIT_SHA_RE.match(sha):
+            return sha
+    return None
+
+
+def read_git_commit_sha_via_subprocess(path: str | Path) -> str:
+    """Fall back to `git rev-parse HEAD` for unusual repository layouts.
+
+    Args:
+        path: Directory or file path inside a repository.
+
+    Returns:
+        The full commit SHA reported by git, or an empty string on failure.
+    """
+    import subprocess  # noqa: S404  # stdlib subprocess fallback
+
+    try:
+        result = subprocess.run(
+            ["git", "rev-parse", "HEAD"],  # noqa: S607
+            capture_output=True,
+            text=True,
+            timeout=2,
+            check=False,
+            cwd=path,
+        )
+        if result.returncode == 0:
+            return result.stdout.strip()
+    except FileNotFoundError:
+        pass  # git not installed
+    except subprocess.TimeoutExpired:
+        logger.debug("Git commit detection timed out")
+    except OSError:
+        logger.debug("Git commit detection failed", exc_info=True)
+    return ""
+
+
+def resolve_git_commit_sha(path: str | Path) -> str:
+    """Resolve the current `HEAD` commit SHA, filesystem-first.
+
+    Args:
+        path: Directory or file path inside a repository.
+
+    Returns:
+        The full commit SHA, or an empty string when none can be determined.
+    """
+    sha = read_git_commit_sha_from_filesystem(path)
+    if sha is not None:
+        return sha
+    return read_git_commit_sha_via_subprocess(path)
+
+
+def read_git_remote_url_from_filesystem(path: str | Path) -> str | None:
+    """Read the `origin` remote URL from the repository `config` file.
+
+    Args:
+        path: Directory or file path inside a repository.
+
+    Returns:
+        The `origin` remote URL, an empty string when `path` is not inside a
+        git repository, or `None` when no `origin` URL is configured.
+    """
+    git_dir = find_git_dir(path)
+    if git_dir is None:
+        return ""
+
+    try:
+        raw = (git_dir / "config").read_text(encoding="utf-8")
+    except FileNotFoundError:
+        return None
+    except OSError:
+        logger.debug("Failed to read git config in %s", git_dir, exc_info=True)
+        return None
+
+    in_origin = False
+    for line in raw.splitlines():
+        stripped = line.strip()
+        if stripped.startswith("[") and stripped.endswith("]"):
+            # Section header, e.g. [remote "origin"]. Match case-insensitively
+            # on the remote name to mirror git's own behavior.
+            in_origin = stripped.replace(" ", "").lower() == '[remote"origin"]'
+            continue
+        if in_origin and stripped.lower().startswith("url"):
+            _, _, value = stripped.partition("=")
+            url = value.strip()
+            if url:
+                return url
+    return None
+
+
+def read_git_remote_url_via_subprocess(path: str | Path) -> str:
+    """Fall back to `git config --get remote.origin.url`.
+
+    Args:
+        path: Directory or file path inside a repository.
+
+    Returns:
+        The `origin` remote URL reported by git, or an empty string on failure.
+    """
+    import subprocess  # noqa: S404  # stdlib subprocess fallback
+
+    try:
+        result = subprocess.run(
+            ["git", "config", "--get", "remote.origin.url"],  # noqa: S607
+            capture_output=True,
+            text=True,
+            timeout=2,
+            check=False,
+            cwd=path,
+        )
+        if result.returncode == 0:
+            return result.stdout.strip()
+    except FileNotFoundError:
+        pass  # git not installed
+    except subprocess.TimeoutExpired:
+        logger.debug("Git remote detection timed out")
+    except OSError:
+        logger.debug("Git remote detection failed", exc_info=True)
+    return ""
+
+
+def resolve_git_remote_url(path: str | Path) -> str:
+    """Resolve the `origin` remote URL, filesystem-first.
+
+    Args:
+        path: Directory or file path inside a repository.
+
+    Returns:
+        The `origin` remote URL, or an empty string when none can be determined.
+    """
+    url = read_git_remote_url_from_filesystem(path)
+    if url is not None:
+        return url
+    return read_git_remote_url_via_subprocess(path)
+
+
+_REPO_PROVIDERS: dict[str, str] = {
+    "github.com": "github",
+    "gitlab.com": "gitlab",
+    "bitbucket.org": "bitbucket",
+}
+"""Maps a known git host to its `repository_provider` slug."""
+
+
+class RepositoryMetadata(NamedTuple):
+    """Parsed `origin` remote attribution for coding-agent-v1 traces.
+
+    A `NamedTuple` so callers can still unpack positionally or index, while the
+    field names keep the slot order from being load-bearing at every call site.
+    """
+
+    url: str
+    """Normalized `https://<host>/<org>/<repo>` URL (credentials stripped)."""
+
+    provider: str
+    """Provider slug: `github`, `gitlab`, `bitbucket`, or `other`."""
+
+    name: str
+    """`org/repo` name (may be nested, e.g. `group/subgroup/project`)."""
+
+
+def parse_repository_metadata(remote_url: str) -> RepositoryMetadata | None:
+    """Derive repository attribution from an `origin` remote URL.
+
+    Handles both HTTPS (`https://github.com/org/repo.git`) and scp-style SSH
+    (`git@github.com:org/repo.git`) remotes, normalizing the URL to its
+    canonical `https://<host>/<org>/<repo>` form and stripping any embedded
+    credentials.
+
+    Args:
+        remote_url: The raw `origin` remote URL.
+
+    Returns:
+        A `RepositoryMetadata` of the normalized URL, provider slug
+            (`github`/`gitlab`/`bitbucket`/`other`), and `org/repo` name,
+            or `None` when the URL cannot be parsed.
+    """
+    url = (remote_url or "").strip()
+    if not url:
+        return None
+
+    host: str
+    repo_path: str
+    if "://" in url:
+        parsed = urlparse(url)
+        host = (parsed.hostname or "").lower()
+        repo_path = parsed.path.lstrip("/")
+    else:
+        # scp-style: [user@]host:org/repo(.git)
+        userhost, sep, repo_path = url.partition(":")
+        if not sep:
+            return None
+        host = userhost.rsplit("@", 1)[-1].lower()
+        repo_path = repo_path.lstrip("/")
+
+    repo_path = repo_path.strip("/").removesuffix(".git").strip("/")
+    if not host or not repo_path:
+        return None
+
+    provider = _REPO_PROVIDERS.get(host, "other")
+    normalized_url = f"https://{host}/{repo_path}"
+    return RepositoryMetadata(normalized_url, provider, repo_path)

libs/code/deepagents_code/app.py

@@ -1486,8 +1486,43 @@ class TextualSessionState:
             thread_id: Optional thread ID (generates UUID7 if not provided)
         """
         self.auto_approve = auto_approve
-        self.thread_id = thread_id or _new_thread_id()
         self.approval_mode_key: str | None = None
+        self.turn_number = 0
+        """1-based user-turn count for the thread (coding-agent-v1 turn_number)."""
+        self.turn_id: str | None = None
+        """Stable id for the current user turn (coding-agent-v1 turn_id)."""
+        # Assign the backing field directly: the setter reads `self._thread_id`
+        # to detect a thread change, and it isn't set yet.
+        self._thread_id = thread_id or _new_thread_id()
+
+    @property
+    def thread_id(self) -> str:
+        """Active LangGraph thread id for the session."""
+        return self._thread_id
+
+    @thread_id.setter
+    def thread_id(self, value: str) -> None:
+        # Per-thread turn markers (coding-agent-v1): restart on every thread
+        # change so traces never inherit the prior thread's sequence.
+        if value != self._thread_id:
+            self.turn_number = 0
+            self.turn_id = None
+        self._thread_id = value
+
+    def advance_turn(self) -> tuple[str, int]:
+        """Begin a new user turn, advancing the per-thread turn markers.
+
+        Generates a fresh `turn_id` and increments `turn_number`. Call once per
+        user prompt, before building the stream config.
+
+        Returns:
+            The `(turn_id, turn_number)` for the new turn.
+        """
+        from uuid import uuid4
+
+        self.turn_number += 1
+        self.turn_id = str(uuid4())
+        return self.turn_id, self.turn_number
 
     def reset_thread(self) -> str:
         """Reset to a new thread.
@@ -1495,7 +1530,7 @@ class TextualSessionState:
         Returns:
             The new thread_id.
         """
-        self.thread_id = _new_thread_id()
+        self.thread_id = _new_thread_id()  # setter resets the turn markers
         self.approval_mode_key = None
         return self.thread_id
 

libs/code/deepagents_code/config.py

@@ -790,6 +790,8 @@ if TYPE_CHECKING:
     from langchain_core.runnables import RunnableConfig
     from rich.console import Console
 
+    from deepagents_code._git import RepositoryMetadata
+
     # Static type stubs for lazy module attributes resolved by __getattr__.
     # At runtime these are created on first access by _get_settings() /
     # _get_console() and cached in globals().
@@ -1244,33 +1246,201 @@ def _get_git_branch() -> str | None:
     return branch
 
 
+_repo_metadata_cache: dict[str, RepositoryMetadata | None] = {}
+"""Per-cwd cache of resolved repository metadata."""
+
+
+def _get_git_commit_sha() -> str | None:
+    """Return the current `HEAD` commit SHA, or `None` if unavailable.
+
+    Resolved fresh on every call (unlike the branch/repo lookups): `HEAD` moves
+    whenever the agent or user commits, checks out, or resets within a session,
+    and each turn's trace must record the commit that was current for that turn.
+    """
+    from deepagents_code._git import resolve_git_commit_sha
+
+    try:
+        cwd = str(Path.cwd())
+    except OSError:
+        logger.debug("Could not determine cwd for git commit lookup", exc_info=True)
+        return None
+
+    try:
+        return resolve_git_commit_sha(cwd) or None
+    except OSError:
+        logger.debug("Could not determine git commit", exc_info=True)
+        return None
+
+
+def _get_repository_metadata() -> RepositoryMetadata | None:
+    """Return parsed `origin` repository metadata, or `None`."""
+    from deepagents_code._git import parse_repository_metadata, resolve_git_remote_url
+
+    try:
+        cwd = str(Path.cwd())
+    except OSError:
+        logger.debug("Could not determine cwd for git remote lookup", exc_info=True)
+        return None
+    if cwd in _repo_metadata_cache:
+        return _repo_metadata_cache[cwd]
+
+    repo: RepositoryMetadata | None = None
+    try:
+        remote_url = resolve_git_remote_url(cwd)
+        if remote_url:
+            repo = parse_repository_metadata(remote_url)
+    except OSError:
+        logger.debug("Could not determine git remote", exc_info=True)
+
+    _repo_metadata_cache[cwd] = repo
+    return repo
+
+
+# coding-agent-v1 contract literals (LSEN-277). See `build_coding_agent_metadata`.
+CODING_AGENT_KIND = "coding_agent"
+"""Fixed `ls_agent_kind` literal identifying the coding-agent trace class."""
+
+CODING_AGENT_INTEGRATION = "deepagents-code"
+"""Stable `ls_integration` id for this plugin (unchanged for backward-compat)."""
+
+CODING_AGENT_RUNTIME = "Deep Agents Code"
+"""User-facing `ls_agent_runtime` name."""
+
+CODING_AGENT_TRACE_SCHEMA_VERSION = "coding-agent-v1"
+"""Version of the coding-agent trace-metadata contract this build emits."""
+
+
+def build_coding_agent_metadata(
+    *,
+    thread_id: str,
+    turn_id: str | None,
+    turn_number: int | None,
+    cwd: str,
+    git_branch: str | None,
+    sandbox_type: str | None,
+    user_id: str | None,
+) -> dict[str, Any]:
+    """Build the shared coding-agent-v1 trace-metadata block.
+
+    Implements the `coding-agent-v1` contract (LSEN-277) for Deep Agents Code:
+    one helper that stamps the identity block, plugin/runtime versions, turn
+    markers, and repo/git/cwd attribution. The seven identity/version keys and
+    `thread_id` are always present; the optional keys whose value is unknown are
+    omitted (per the contract), so callers can pass `None` for any of them.
+
+    Because Deep Agents Code is itself the runtime — there is no separate CLI
+    package — `ls_integration_version` and `ls_agent_runtime_version` both come
+    from the `deepagents-code` package version (`__version__`). The underlying
+    `deepagents` SDK version is surfaced separately as
+    `dcode_client_deepagents_version` by `build_stream_config`.
+
+    Scope-restricted contract keys are intentionally NOT produced here:
+    `approval_policy` (root/interrupted only) and `ls_subagent_id` /
+    `ls_subagent_type` (subagent only). This metadata propagates trace-wide
+    through the LangGraph stream config (and, for subagents, the per-key config
+    merge of langgraph#7926 / deepagents#3634), so any key placed here lands on
+    every descendant run. Emitting a run-type-scoped key would therefore leak it
+    onto run types outside its contract `appliesTo` set — a hard validator
+    failure — and the LangGraph runtime exposes no clean per-run-type metadata
+    seam to scope them. See `build_stream_config` for the full rationale.
+
+    Args:
+        thread_id: Stable conversation id; also set as top-level `thread_id`.
+        turn_id: Per-turn id (uuid4 / message id), or `None`.
+        turn_number: 1-based per-thread turn index, or `None`.
+        cwd: Current working directory, or empty string when unavailable.
+        git_branch: Current branch name, or `None`.
+        sandbox_type: Sandbox provider name, or `None`/`"none"` when inactive.
+        user_id: Stable pseudonymous user id, or `None`.
+
+    Returns:
+        The contract metadata dict with unknown keys omitted.
+    """
+    metadata: dict[str, Any] = {
+        "ls_agent_kind": CODING_AGENT_KIND,
+        "ls_integration": CODING_AGENT_INTEGRATION,
+        "ls_agent_runtime": CODING_AGENT_RUNTIME,
+        "thread_id": thread_id,
+        "ls_trace_schema_version": CODING_AGENT_TRACE_SCHEMA_VERSION,
+        "ls_integration_version": __version__,
+        "ls_agent_runtime_version": __version__,
+    }
+
+    if turn_id:
+        metadata["turn_id"] = turn_id
+    if turn_number is not None:
+        metadata["turn_number"] = turn_number
+
+    repo = _get_repository_metadata()
+    if repo is not None:
+        repository_url, repository_provider, repository_name = repo
+        metadata["repository_url"] = repository_url
+        metadata["repository_provider"] = repository_provider
+        metadata["repository_name"] = repository_name
+
+    if git_branch:
+        metadata["git_branch"] = git_branch
+    commit_sha = _get_git_commit_sha()
+    if commit_sha:
+        metadata["git_commit_sha"] = commit_sha
+    if cwd:
+        metadata["cwd"] = cwd
+
+    if user_id:
+        metadata["user_id"] = user_id
+    if sandbox_type and sandbox_type != "none":
+        metadata["sandbox_type"] = sandbox_type
+
+    return metadata
+
+
 def build_stream_config(
     thread_id: str,
     assistant_id: str | None,
     *,
     sandbox_type: str | None = None,
+    turn_id: str | None = None,
+    turn_number: int | None = None,
 ) -> RunnableConfig:
     """Build the LangGraph stream config dict.
 
-    Injects the dcode version into `metadata["lc_versions"]` so LangSmith traces
-    can be correlated with specific releases. `create_deep_agent` supplies the
-    SDK version through the compiled graph config, and LangChain merges nested
-    metadata dictionaries so both versions survive at stream time.
+    Stamps the shared `coding-agent-v1` trace-metadata contract (LSEN-277) via
+    `build_coding_agent_metadata` — identity block, plugin/runtime versions,
+    turn markers, and repo/git/cwd attribution — onto `metadata`. Metadata set
+    here propagates trace-wide to every run in the graph (root, llm, tool, and
+    subagent subgraphs), which is exactly what the contract's "always" and
+    "where-known" keys require, so the helper output is stamped once here.
+
+    Scope-restricted contract keys are deliberately not emitted. `approval_policy`
+    (root/interrupted only) and `ls_subagent_id` / `ls_subagent_type` (subagent
+    only) cannot live in this trace-wide metadata: LangGraph propagates each key
+    to all descendant runs (per-key config merge, langgraph#7926 /
+    deepagents#3634), so they would leak onto run types outside their contract
+    `appliesTo` set and fail validation. This runtime exposes no clean
+    per-run-type metadata seam to scope them, so they are omitted by design
+    rather than leaked. (Subagent runs still inherit the parent/root `thread_id`
+    and all required keys, satisfying the contract's grouping rule.)
+
+    Also injects the dcode version into `metadata["lc_versions"]` so LangSmith
+    traces can be correlated with specific releases. `create_deep_agent` supplies
+    the SDK version through the compiled graph config, and LangChain merges
+    nested metadata dictionaries so both versions survive at stream time.
 
     Also records `dcode_client_deepagents_version` as a dcode-client diagnostic.
     This describes the Deep Agents package installed alongside the TUI, which
     can differ from a remote graph's Deep Agents runtime version.
 
-    Includes `ls_integration` metadata so LangSmith traces originating from
-    the app are distinguishable from bare SDK usage.
-
     Args:
-        thread_id: The app session thread identifier.
+        thread_id: The app session thread identifier. Set both on
+            `configurable.thread_id` and as the top-level `metadata.thread_id`
+            used by the contract for grouping turns.
         assistant_id: The dcode agent identifier, if any. When set, it is
             surfaced in trace metadata under `dcode_agent_name` and
             `agent_name`.
         sandbox_type: Sandbox provider name for trace metadata, or `None` if no
             sandbox is active.
+        turn_id: Stable per-turn id for the current user prompt, or `None`.
+        turn_number: 1-based per-thread turn index, or `None`.
 
     Returns:
         Config dict with `configurable` and `metadata` keys.
@@ -1283,21 +1453,24 @@ def build_stream_config(
         logger.warning("Could not determine working directory", exc_info=True)
         cwd = ""
 
-    metadata: dict[str, Any] = {
-        "lc_versions": {"deepagents-code": __version__},
-        "ls_integration": "deepagents-code",
-    }
+    from deepagents_code._env_vars import USER_ID
+
+    metadata: dict[str, Any] = build_coding_agent_metadata(
+        thread_id=thread_id,
+        turn_id=turn_id,
+        turn_number=turn_number,
+        cwd=cwd,
+        git_branch=_get_git_branch(),
+        sandbox_type=sandbox_type,
+        user_id=os.environ.get(USER_ID) or None,
+    )
+
+    # Legacy / diagnostic keys preserved for backward-compatibility during the
+    # coding-agent-v1 rollout (not part of the contract).
+    metadata["lc_versions"] = {"deepagents-code": __version__}
     deepagents_version = _get_deepagents_version()
     if deepagents_version is not None:
         metadata["dcode_client_deepagents_version"] = deepagents_version
-
-    from deepagents_code._env_vars import USER_ID
-
-    user_id = os.environ.get(USER_ID)
-    if user_id:
-        metadata["user_id"] = user_id
-    if cwd:
-        metadata["cwd"] = cwd
     if assistant_id:
         metadata.update(
             {
@@ -1306,11 +1479,7 @@ def build_stream_config(
                 "updated_at": datetime.now(UTC).isoformat(),
             }
         )
-    branch = _get_git_branch()
-    if branch:
-        metadata["git_branch"] = branch
-    if sandbox_type and sandbox_type != "none":
-        metadata["sandbox_type"] = sandbox_type
+
     return {
         "configurable": {"thread_id": thread_id},
         "metadata": metadata,
@@ -2670,18 +2839,25 @@ def configure_langsmith_secret_redaction() -> bool:
 
     env = dict(os.environ)
     # Cheap env-var checks first so the common (tracing-off) startup path skips
-    # the TOML read in `is_langsmith_redaction_enabled`.
+    # the TOML read in `is_langsmith_redaction_enabled`. These are plain env
+    # reads with no failure mode of their own, so they stay outside the
+    # fail-closed boundary: if there is no upload target, there is nothing to
+    # protect.
     if not (_tracing_enabled_from(env) and _tracing_can_upload_from(env)):
         return False
-    if not is_langsmith_redaction_enabled():
-        logger.warning(
-            "LangSmith tracing is active but secret redaction is disabled via "
-            "%s; secrets may be uploaded to traces unredacted.",
-            LANGSMITH_REDACT,
-        )
-        return False
 
+    # Everything from here on runs inside the fail-closed boundary: any
+    # unexpected exception (including from the redaction-toggle lookup) disables
+    # tracing rather than escaping and leaving tracing live but unredacted.
     try:
+        if not is_langsmith_redaction_enabled():
+            logger.warning(
+                "LangSmith tracing is active but secret redaction is disabled "
+                "via %s; secrets may be uploaded to traces unredacted.",
+                LANGSMITH_REDACT,
+            )
+            return False
+
         from langsmith import Client, configure
         from langsmith.anonymizer import create_secret_anonymizer
 
@@ -2715,11 +2891,16 @@ def configure_langsmith_secret_redaction() -> bool:
 def _fail_closed_disable_tracing() -> None:
     """Best-effort disable LangSmith tracing after a redaction setup failure.
 
-    Tries the SDK's global tracing switch first. If even that call fails, the
-    canonical tracing-enable env vars (and their `DEEPAGENTS_CODE_`-prefixed
-    forms) are cleared as a last-resort barrier: the LangChain tracer consults
-    the global switch first but falls back to these env vars, so removing them
-    prevents an unredacted upload when the global switch could not be set.
+    The SDK's global tracing switch (`configure(enabled=False)`) is the primary,
+    load-bearing control and is tried first. Clearing the canonical
+    tracing-enable env vars (and their `DEEPAGENTS_CODE_`-prefixed forms) is only
+    a last-resort fallback for the case where even that call fails (e.g. the
+    `langsmith` import is broken): the LangChain tracer checks the global switch
+    first but falls back to these env vars, so removing them helps prevent a
+    newly created tracer from starting an unredacted upload. (It only helps —
+    the SDK's env-var lookup is `lru_cache`d, so a value already read this
+    process may still be served from cache; the global switch is the reliable
+    stop.)
     """
     try:
         from langsmith import configure

libs/code/deepagents_code/non_interactive.py

@@ -1217,10 +1217,17 @@ async def run_non_interactive(
 
     thread_id = generate_thread_id()
 
+    # One user turn per process: fresh turn id, turn_number 1.
+    from uuid import uuid4
+
     from deepagents_code.config import build_stream_config
 
     config: RunnableConfig = build_stream_config(
-        thread_id, assistant_id, sandbox_type=sandbox_type
+        thread_id,
+        assistant_id,
+        sandbox_type=sandbox_type,
+        turn_id=str(uuid4()),
+        turn_number=1,
     )
 
     thread_url_lookup: ThreadUrlLookupState | None = None

libs/code/deepagents_code/textual_adapter.py

@@ -565,7 +565,27 @@ async def execute_task_textual(
         message_content = final_input
 
     thread_id = session_state.thread_id
-    config = build_stream_config(thread_id, assistant_id, sandbox_type=sandbox_type)
+    # Advance the per-thread turn markers (coding-agent-v1 turn_id/turn_number)
+    # once per user prompt, before building the stream config. `session_state`
+    # is duck-typed (`Any`): the production `TextualSessionState` always has
+    # `advance_turn`, but lightweight callers/test doubles may not, so probe for
+    # it and degrade to no turn markers rather than raising.
+    advance_turn = getattr(session_state, "advance_turn", None)
+    if callable(advance_turn):
+        turn_id, turn_number = advance_turn()
+    else:
+        turn_id, turn_number = None, None
+    # `build_stream_config` does blocking git filesystem reads and may shell out
+    # to `git`; offload it so the Textual event loop stays responsive. Advancing
+    # the turn markers above is pure/cheap and stays on the loop.
+    config = await asyncio.to_thread(
+        build_stream_config,
+        thread_id,
+        assistant_id,
+        sandbox_type=sandbox_type,
+        turn_id=turn_id,
+        turn_number=turn_number,
+    )
 
     await dispatch_hook("session.start", {"thread_id": thread_id})
 

libs/code/tests/unit_tests/data/coding_agent_v1_validator.json

@@ -0,0 +1,241 @@
+{
+  "schemaVersion": "coding-agent-v1",
+  "description": "Machine-readable contract for LangSmith coding-agent trace metadata. Set on run.extra.metadata. 'requirement' tiers: always = on every run; where_known = required when the runtime exposes the value, omit otherwise; contextual = optional. runTypes: root | llm | tool | subagent | interrupted.",
+  "integrations": ["claude-code", "openai-codex", "deepagents-code", "cursor", "pi"],
+  "runtimeNames": {
+    "claude-code": "Claude Code",
+    "openai-codex": "Codex",
+    "deepagents-code": "Deep Agents Code",
+    "cursor": "Cursor",
+    "pi": "Pi"
+  },
+  "keys": [
+    {
+      "key": "ls_agent_kind",
+      "appliesTo": ["root", "llm", "tool", "subagent", "interrupted"],
+      "type": "string",
+      "allowedValues": ["coding_agent"],
+      "requirement": "always",
+      "requiredWhereKnown": false
+    },
+    {
+      "key": "ls_integration",
+      "appliesTo": ["root", "llm", "tool", "subagent", "interrupted"],
+      "type": "string",
+      "allowedValues": ["claude-code", "openai-codex", "deepagents-code", "cursor", "pi"],
+      "requirement": "always",
+      "requiredWhereKnown": false
+    },
+    {
+      "key": "ls_agent_runtime",
+      "appliesTo": ["root", "llm", "tool", "subagent", "interrupted"],
+      "type": "string",
+      "allowedValues": ["Claude Code", "Codex", "Deep Agents Code", "Cursor", "Pi"],
+      "requirement": "always",
+      "requiredWhereKnown": false
+    },
+    {
+      "key": "thread_id",
+      "appliesTo": ["root", "llm", "tool", "subagent", "interrupted"],
+      "type": "string",
+      "allowedValues": null,
+      "requirement": "always",
+      "requiredWhereKnown": false
+    },
+    {
+      "key": "ls_trace_schema_version",
+      "appliesTo": ["root", "llm", "tool", "subagent", "interrupted"],
+      "type": "string",
+      "allowedValues": ["coding-agent-v1"],
+      "requirement": "always",
+      "requiredWhereKnown": false
+    },
+    {
+      "key": "ls_integration_version",
+      "appliesTo": ["root", "llm", "tool", "subagent", "interrupted"],
+      "type": "string",
+      "allowedValues": null,
+      "requirement": "where_known",
+      "requiredWhereKnown": true,
+      "note": "Bundled plugins (cursor, claude-code, pi) must inject at build time via esbuild define; do not require('./package.json')."
+    },
+    {
+      "key": "ls_agent_runtime_version",
+      "appliesTo": ["root", "llm", "tool", "subagent", "interrupted"],
+      "type": "string",
+      "allowedValues": null,
+      "requirement": "where_known",
+      "requiredWhereKnown": true
+    },
+    {
+      "key": "turn_id",
+      "appliesTo": ["root", "llm", "tool", "subagent", "interrupted"],
+      "type": "string",
+      "allowedValues": null,
+      "requirement": "where_known",
+      "requiredWhereKnown": true,
+      "note": "At least one of turn_id / turn_number is required wherever the runtime exposes turns."
+    },
+    {
+      "key": "turn_number",
+      "appliesTo": ["root", "llm", "tool", "subagent", "interrupted"],
+      "type": "integer",
+      "allowedValues": null,
+      "requirement": "where_known",
+      "requiredWhereKnown": true,
+      "note": "1-based. Pi source is 0-based turnIndex -> emit turnIndex + 1."
+    },
+    {
+      "key": "repository_url",
+      "appliesTo": ["root", "llm", "tool", "subagent", "interrupted"],
+      "type": "string",
+      "allowedValues": null,
+      "requirement": "where_known",
+      "requiredWhereKnown": true
+    },
+    {
+      "key": "repository_provider",
+      "appliesTo": ["root", "llm", "tool", "subagent", "interrupted"],
+      "type": "string",
+      "allowedValues": null,
+      "suggestedValues": ["github", "gitlab", "bitbucket", "other"],
+      "requirement": "where_known",
+      "requiredWhereKnown": true
+    },
+    {
+      "key": "repository_name",
+      "appliesTo": ["root", "llm", "tool", "subagent", "interrupted"],
+      "type": "string",
+      "allowedValues": null,
+      "requirement": "where_known",
+      "requiredWhereKnown": true
+    },
+    {
+      "key": "git_branch",
+      "appliesTo": ["root", "llm", "tool", "subagent", "interrupted"],
+      "type": "string",
+      "allowedValues": null,
+      "requirement": "where_known",
+      "requiredWhereKnown": true
+    },
+    {
+      "key": "git_commit_sha",
+      "appliesTo": ["root", "llm", "tool", "subagent", "interrupted"],
+      "type": "string",
+      "allowedValues": null,
+      "requirement": "where_known",
+      "requiredWhereKnown": true
+    },
+    {
+      "key": "cwd",
+      "appliesTo": ["root", "llm", "tool", "subagent", "interrupted"],
+      "type": "string",
+      "allowedValues": null,
+      "requirement": "where_known",
+      "requiredWhereKnown": true
+    },
+    {
+      "key": "user_id",
+      "appliesTo": ["root", "llm", "tool", "subagent", "interrupted"],
+      "type": "string",
+      "allowedValues": null,
+      "requirement": "contextual",
+      "requiredWhereKnown": false,
+      "note": "Stable, pseudonymous; preferred over local_username."
+    },
+    {
+      "key": "local_username",
+      "appliesTo": ["root", "llm", "tool", "subagent", "interrupted"],
+      "type": "string",
+      "allowedValues": null,
+      "requirement": "contextual",
+      "requiredWhereKnown": false,
+      "note": "PII-sensitive, backward-compatible."
+    },
+    {
+      "key": "user_email",
+      "appliesTo": ["root", "llm", "tool", "subagent", "interrupted"],
+      "type": "string",
+      "allowedValues": null,
+      "requirement": "contextual",
+      "requiredWhereKnown": false,
+      "note": "Provisional / PII. Emit where known, omit otherwise; may be scrapped later."
+    },
+    {
+      "key": "sandbox_type",
+      "appliesTo": ["root", "llm", "tool", "subagent", "interrupted"],
+      "type": "string",
+      "allowedValues": null,
+      "requirement": "contextual",
+      "requiredWhereKnown": false
+    },
+    {
+      "key": "approval_policy",
+      "appliesTo": ["root", "interrupted"],
+      "type": "string",
+      "allowedValues": null,
+      "requirement": "contextual",
+      "requiredWhereKnown": false,
+      "note": "Permission mode / policy for the turn. Root + interrupted only."
+    },
+    {
+      "key": "ls_subagent_id",
+      "appliesTo": ["subagent"],
+      "type": "string",
+      "allowedValues": null,
+      "requirement": "contextual",
+      "requiredWhereKnown": false
+    },
+    {
+      "key": "ls_subagent_type",
+      "appliesTo": ["subagent"],
+      "type": "string",
+      "allowedValues": null,
+      "requirement": "contextual",
+      "requiredWhereKnown": false,
+      "note": "Subagent runs only. Do NOT emit on root runs (e.g. do not copy claude-code's old ls_agent_type='root' here)."
+    },
+    {
+      "key": "ls_tool_name",
+      "appliesTo": ["tool"],
+      "type": "string",
+      "allowedValues": null,
+      "requirement": "contextual",
+      "requiredWhereKnown": false,
+      "note": "Tool runs only; emit when it differs from the run name. Pass-through native name — no cross-agent taxonomy normalization."
+    }
+  ],
+  "preserveExistingOnModelAndToolRuns": [
+    "ls_provider",
+    "ls_model_name",
+    "ls_invocation_params",
+    "usage_metadata"
+  ],
+  "perIntegrationSources": {
+    "claude-code": {
+      "thread_id": "session_id (hook payload)",
+      "turn_id": "promptId (transcript, native)",
+      "turn_number": "turn_count"
+    },
+    "openai-codex": {
+      "thread_id": "thread_id",
+      "turn_id": "turn_id",
+      "turn_number": "native"
+    },
+    "cursor": {
+      "thread_id": "conversation_id",
+      "turn_id": "generation_id",
+      "turn_number": "turnNum"
+    },
+    "pi": {
+      "thread_id": "ctx.sessionManager.getSessionId()",
+      "turn_id": "none -> use turnIndex as turn_number",
+      "turn_number": "turnIndex (0-based) + 1"
+    },
+    "deepagents-code": {
+      "thread_id": "configurable.thread_id",
+      "turn_id": "add — uuid4 / msg-id per prompt",
+      "turn_number": "add — counter on session_state"
+    }
+  }
+}

libs/code/tests/unit_tests/test_coding_agent_metadata.py

@@ -0,0 +1,255 @@
+"""Contract tests for the coding-agent-v1 trace-metadata standard (LSEN-277).
+
+These load the machine-readable contract (`validator.json`, vendored under
+`data/`) and assert that the metadata Deep Agents Code stamps onto its
+LangGraph stream config satisfies the contract — required keys, types,
+allowed values, and the run-type `appliesTo` rules — for every run type the
+trace-wide metadata block lands on.
+
+The vendored validator is a copy of the shared
+`coding-agent-v1/validator.json`; keep it in sync when the contract changes.
+End-to-end acceptance is a live trace validated with `validate-thread.mjs`
+(the `deepagents-code` profile), not these hermetic unit tests.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import TYPE_CHECKING
+from unittest.mock import patch
+
+import pytest
+
+from deepagents_code import config as config_module
+from deepagents_code._version import __version__
+from deepagents_code.config import build_coding_agent_metadata, build_stream_config
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+_VALIDATOR_PATH = Path(__file__).parent / "data" / "coding_agent_v1_validator.json"
+
+# Run types the trace-wide stream-config metadata propagates to.
+_TRACE_WIDE_RUN_TYPES = ("root", "llm", "tool", "subagent", "interrupted")
+
+# Scope-restricted keys not emitted (would leak trace-wide; see helper docstring).
+_OMITTED_SCOPE_RESTRICTED_KEYS = frozenset(
+    {"approval_policy", "ls_subagent_id", "ls_subagent_type"}
+)
+
+
+@pytest.fixture(scope="module")
+def contract() -> dict:
+    """Load the vendored coding-agent-v1 validator contract."""
+    return json.loads(_VALIDATOR_PATH.read_text(encoding="utf-8"))
+
+
+def _type_ok(value: object, type_name: str) -> bool:
+    if type_name == "string":
+        return isinstance(value, str)
+    if type_name == "integer":
+        return isinstance(value, int) and not isinstance(value, bool)
+    return True
+
+
+def _validate(
+    metadata: dict, run_type: str, contract: dict
+) -> tuple[list[str], list[str]]:
+    """Mirror validate-thread.mjs's per-run rules for one metadata dict.
+
+    This re-implements the external `validate-thread.mjs` (`deepagents-code`
+    profile) rules in Python because that validator lives in another toolchain
+    and can't be imported here; keep this in lock-step with it when the contract
+    changes. It is intentionally a slight over-approximation: it treats each of
+    `turn_id` / `turn_number` as independently `requiredWhereKnown` rather than
+    enforcing the contract's "at least one of" OR-semantics. Deep Agents Code
+    always emits both, so the simplification is sound today; revisit if a run
+    type ever emits only one. End-to-end acceptance is the live `.mjs` run, not
+    this hermetic approximation.
+
+    Returns:
+        `(errors, missing_where_known)` for `metadata` classified as `run_type`.
+    """
+    errors: list[str] = []
+    missing_where_known: list[str] = []
+
+    for spec in contract["keys"]:
+        key = spec["key"]
+        applies = run_type in spec["appliesTo"]
+        present = key in metadata
+
+        if applies and not present:
+            if spec["requirement"] == "always":
+                errors.append(f"missing required key {key!r}")
+            elif spec["requirement"] == "where_known" and spec.get(
+                "requiredWhereKnown"
+            ):
+                missing_where_known.append(key)
+            continue
+
+        if applies and present:
+            value = metadata[key]
+            if not _type_ok(value, spec["type"]):
+                errors.append(f"{key!r} wrong type: {value!r}")
+            allowed = spec.get("allowedValues")
+            if allowed and value not in allowed:
+                errors.append(f"{key!r}={value!r} not in {allowed}")
+
+        # Leakage: a contract key present on a run type outside its appliesTo.
+        if not applies and present:
+            errors.append(
+                f"{key!r} leaked onto {run_type!r} (only {spec['appliesTo']})"
+            )
+
+    return errors, missing_where_known
+
+
+@pytest.fixture
+def known_env() -> Iterator[None]:
+    """Patch git/user lookups so every where-known contract key is resolvable."""
+    from deepagents_code._git import RepositoryMetadata
+
+    repo = RepositoryMetadata(
+        "https://github.com/langchain-ai/deepagents",
+        "github",
+        "langchain-ai/deepagents",
+    )
+    sha = "a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6e7f8a9b0"
+    with (
+        patch.object(config_module, "_get_git_branch", return_value="main"),
+        patch.object(config_module, "_get_git_commit_sha", return_value=sha),
+        patch.object(config_module, "_get_repository_metadata", return_value=repo),
+        patch.dict("os.environ", {"DEEPAGENTS_CODE_USER_ID": "u_test"}),
+    ):
+        yield
+
+
+@pytest.mark.usefixtures("known_env")
+class TestContractCompliance:
+    """The trace-wide metadata block satisfies the contract on every run type."""
+
+    def test_no_hard_errors_on_any_run_type(self, contract: dict) -> None:
+        config = build_stream_config(
+            "thread-123", assistant_id="agent", turn_id="turn-abc", turn_number=2
+        )
+        metadata = config["metadata"]
+        for run_type in _TRACE_WIDE_RUN_TYPES:
+            errors, _ = _validate(metadata, run_type, contract)
+            assert errors == [], f"{run_type}: {errors}"
+
+    def test_all_where_known_keys_present_when_known(self, contract: dict) -> None:
+        config = build_stream_config(
+            "thread-123", assistant_id="agent", turn_id="turn-abc", turn_number=2
+        )
+        metadata = config["metadata"]
+        # With git + user fully resolvable, no where_known key should be missing.
+        for run_type in _TRACE_WIDE_RUN_TYPES:
+            _, missing = _validate(metadata, run_type, contract)
+            assert missing == [], f"{run_type}: missing where-known {missing}"
+
+    def test_scope_restricted_keys_are_omitted(self) -> None:
+        # These would leak trace-wide and fail validation, so they are not
+        # emitted by design (documented limitation).
+        config = build_stream_config(
+            "thread-123", assistant_id="agent", turn_id="turn-abc", turn_number=2
+        )
+        metadata = config["metadata"]
+        for key in _OMITTED_SCOPE_RESTRICTED_KEYS:
+            assert key not in metadata
+
+
+@pytest.mark.usefixtures("known_env")
+class TestContractValueSemantics:
+    """Exact values of the identity block, versions, and derived keys."""
+
+    def test_identity_block(self) -> None:
+        metadata = build_coding_agent_metadata(
+            thread_id="t1",
+            turn_id="turn-1",
+            turn_number=1,
+            cwd="/work",
+            git_branch="main",
+            sandbox_type=None,
+            user_id=None,
+        )
+        assert metadata["ls_agent_kind"] == "coding_agent"
+        assert metadata["ls_integration"] == "deepagents-code"
+        assert metadata["ls_agent_runtime"] == "Deep Agents Code"
+        assert metadata["ls_trace_schema_version"] == "coding-agent-v1"
+        assert metadata["thread_id"] == "t1"
+
+    def test_versions_coincide_with_package_version(self) -> None:
+        metadata = build_coding_agent_metadata(
+            thread_id="t1",
+            turn_id=None,
+            turn_number=None,
+            cwd="",
+            git_branch=None,
+            sandbox_type=None,
+            user_id=None,
+        )
+        assert metadata["ls_integration_version"] == __version__
+        assert metadata["ls_agent_runtime_version"] == __version__
+
+    def test_repository_keys_from_metadata(self) -> None:
+        metadata = build_coding_agent_metadata(
+            thread_id="t1",
+            turn_id=None,
+            turn_number=None,
+            cwd="",
+            git_branch=None,
+            sandbox_type=None,
+            user_id=None,
+        )
+        assert (
+            metadata["repository_url"] == "https://github.com/langchain-ai/deepagents"
+        )
+        assert metadata["repository_provider"] == "github"
+        assert metadata["repository_name"] == "langchain-ai/deepagents"
+
+    def test_turn_markers_present_and_typed(self) -> None:
+        metadata = build_coding_agent_metadata(
+            thread_id="t1",
+            turn_id="turn-xyz",
+            turn_number=3,
+            cwd="",
+            git_branch=None,
+            sandbox_type=None,
+            user_id=None,
+        )
+        assert metadata["turn_id"] == "turn-xyz"
+        assert metadata["turn_number"] == 3
+        assert isinstance(metadata["turn_number"], int)
+
+
+class TestUnknownKeysOmitted:
+    """Keys with unknown values are omitted regardless of environment."""
+
+    def test_unknown_keys_omitted(self) -> None:
+        with (
+            patch.object(config_module, "_get_git_commit_sha", return_value=None),
+            patch.object(config_module, "_get_repository_metadata", return_value=None),
+        ):
+            metadata = build_coding_agent_metadata(
+                thread_id="t1",
+                turn_id=None,
+                turn_number=None,
+                cwd="",
+                git_branch=None,
+                sandbox_type="none",
+                user_id=None,
+            )
+        for absent in (
+            "turn_id",
+            "turn_number",
+            "repository_url",
+            "git_branch",
+            "git_commit_sha",
+            "cwd",
+            "user_id",
+            "sandbox_type",
+        ):
+            assert absent not in metadata
+        assert metadata["ls_agent_kind"] == "coding_agent"
+        assert metadata["thread_id"] == "t1"

libs/code/tests/unit_tests/test_config.py

@@ -2219,16 +2219,25 @@ class TestLangsmithSecretRedaction:
         self,
         monkeypatch: pytest.MonkeyPatch,
     ) -> None:
-        """When the SDK cannot disable tracing, enable env vars are cleared.
+        """When the SDK cannot disable tracing, all enable env vars are cleared.
 
         This is the worst-case fail-closed path: redaction setup raised and the
-        SDK's `configure(enabled=False)` raised too, so the only remaining barrier
-        is removing the env vars the LangChain tracer falls back to.
+        SDK's `configure(enabled=False)` raised too, so the only remaining
+        barrier is removing every env var the LangChain tracer falls back to —
+        both the canonical names and their `DEEPAGENTS_CODE_`-prefixed forms.
         """
         import os
 
+        from deepagents_code.config import _TRACING_ENABLE_ENV_VARS
+        from deepagents_code.model_config import _ENV_PREFIX
+
         monkeypatch.setenv("LANGSMITH_API_KEY", "lsv2_test")
-        monkeypatch.setenv("LANGSMITH_TRACING", "true")
+        enable_vars = [
+            *_TRACING_ENABLE_ENV_VARS,
+            *(f"{_ENV_PREFIX}{var}" for var in _TRACING_ENABLE_ENV_VARS),
+        ]
+        for var in enable_vars:
+            monkeypatch.setenv(var, "true")
 
         with (
             patch("deepagents_code.config_manifest.load_config_toml", return_value={}),
@@ -2237,7 +2246,32 @@ class TestLangsmithSecretRedaction:
         ):
             assert configure_langsmith_secret_redaction() is False
 
-        assert "LANGSMITH_TRACING" not in os.environ
+        for var in enable_vars:
+            assert var not in os.environ, f"{var} should have been cleared"
+
+    def test_fails_closed_when_redaction_toggle_lookup_raises(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        """An unexpected error from the redaction-toggle lookup fails closed.
+
+        `is_langsmith_redaction_enabled()` runs inside the fail-closed boundary,
+        so even an unexpected exception there disables tracing rather than
+        escaping the function and leaving tracing live but unredacted.
+        """
+        monkeypatch.setenv("DEEPAGENTS_CODE_LANGSMITH_API_KEY", "lsv2_test")
+        monkeypatch.setenv("DEEPAGENTS_CODE_LANGSMITH_TRACING", "true")
+
+        with (
+            patch(
+                "deepagents_code.config.is_langsmith_redaction_enabled",
+                side_effect=RuntimeError("boom"),
+            ),
+            patch("langsmith.configure") as configure,
+        ):
+            assert configure_langsmith_secret_redaction() is False
+
+        configure.assert_called_once_with(enabled=False)
 
     def test_reconfigures_on_each_call(
         self,

libs/code/tests/unit_tests/test_git.py

@@ -8,17 +8,28 @@ from unittest.mock import MagicMock, patch
 import pytest
 
 from deepagents_code._git import (
+    RepositoryMetadata,
     _abbreviate_git_ref,
     _git_dir_cache,
     _normalize_lookup_path,
     _parse_git_dir_pointer,
     find_git_dir,
     find_git_root,
+    parse_repository_metadata,
     read_git_branch_from_filesystem,
     read_git_branch_via_subprocess,
+    read_git_commit_sha_from_filesystem,
+    read_git_commit_sha_via_subprocess,
+    read_git_remote_url_from_filesystem,
+    read_git_remote_url_via_subprocess,
     resolve_git_branch,
+    resolve_git_commit_sha,
+    resolve_git_remote_url,
 )
 
+_FULL_SHA = "a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6e7f8a9b0"
+"""A valid 40-char SHA-1 used across the commit-resolution tests."""
+
 
 @pytest.fixture(autouse=True)
 def clear_git_dir_cache() -> Iterator[None]:
@@ -262,3 +273,317 @@ class TestResolveGitBranch:
         mock_sub.return_value = "fallback-branch"
         assert resolve_git_branch("/some/path") == "fallback-branch"
         mock_sub.assert_called_once()
+
+
+class TestReadGitCommitShaFromFilesystem:
+    def test_loose_ref(self, tmp_path: Path) -> None:
+        git_dir = tmp_path / ".git"
+        (git_dir / "refs" / "heads").mkdir(parents=True)
+        (git_dir / "HEAD").write_text("ref: refs/heads/main\n")
+        (git_dir / "refs" / "heads" / "main").write_text(f"{_FULL_SHA}\n")
+        assert read_git_commit_sha_from_filesystem(tmp_path) == _FULL_SHA
+
+    def test_detached_head(self, tmp_path: Path) -> None:
+        git_dir = tmp_path / ".git"
+        git_dir.mkdir()
+        (git_dir / "HEAD").write_text(f"{_FULL_SHA}\n")
+        assert read_git_commit_sha_from_filesystem(tmp_path) == _FULL_SHA
+
+    def test_packed_ref(self, tmp_path: Path) -> None:
+        git_dir = tmp_path / ".git"
+        git_dir.mkdir()
+        (git_dir / "HEAD").write_text("ref: refs/heads/main\n")
+        (git_dir / "packed-refs").write_text(
+            "# pack-refs with: peeled fully-peeled sorted\n"
+            f"{_FULL_SHA} refs/heads/main\n"
+        )
+        assert read_git_commit_sha_from_filesystem(tmp_path) == _FULL_SHA
+
+    def test_packed_ref_skips_peeled_tag_line(self, tmp_path: Path) -> None:
+        # A `^`-peeled line (the tag's target commit) precedes the wanted ref;
+        # it must be skipped so the ref's own SHA is returned, not the peel.
+        peeled = "0000000000000000000000000000000000000000"
+        git_dir = tmp_path / ".git"
+        git_dir.mkdir()
+        (git_dir / "HEAD").write_text("ref: refs/heads/main\n")
+        (git_dir / "packed-refs").write_text(
+            "# pack-refs with: peeled fully-peeled sorted\n"
+            "1111111111111111111111111111111111111111 refs/tags/v1\n"
+            f"^{peeled}\n"
+            f"{_FULL_SHA} refs/heads/main\n"
+        )
+        assert read_git_commit_sha_from_filesystem(tmp_path) == _FULL_SHA
+
+    def test_not_in_repo_returns_empty(self, tmp_path: Path) -> None:
+        # Empty string (not None) is the not-in-repo signal so resolve() can
+        # short-circuit the subprocess fallback.
+        assert read_git_commit_sha_from_filesystem(tmp_path) == ""
+
+    def test_unresolvable_ref_returns_none(self, tmp_path: Path) -> None:
+        git_dir = tmp_path / ".git"
+        git_dir.mkdir()
+        (git_dir / "HEAD").write_text("ref: refs/heads/missing\n")
+        assert read_git_commit_sha_from_filesystem(tmp_path) is None
+
+
+class TestResolveGitCommitSha:
+    @patch("deepagents_code._git.read_git_commit_sha_via_subprocess")
+    @patch("deepagents_code._git.read_git_commit_sha_from_filesystem")
+    def test_filesystem_wins(self, mock_fs: MagicMock, mock_sub: MagicMock) -> None:
+        mock_fs.return_value = _FULL_SHA
+        assert resolve_git_commit_sha("/some/path") == _FULL_SHA
+        mock_sub.assert_not_called()
+
+    @patch("deepagents_code._git.read_git_commit_sha_via_subprocess")
+    @patch("deepagents_code._git.read_git_commit_sha_from_filesystem")
+    def test_empty_string_skips_subprocess(
+        self, mock_fs: MagicMock, mock_sub: MagicMock
+    ) -> None:
+        mock_fs.return_value = ""
+        assert resolve_git_commit_sha("/some/path") == ""
+        mock_sub.assert_not_called()
+
+    @patch("deepagents_code._git.read_git_commit_sha_via_subprocess")
+    @patch("deepagents_code._git.read_git_commit_sha_from_filesystem")
+    def test_fallback_on_none(self, mock_fs: MagicMock, mock_sub: MagicMock) -> None:
+        mock_fs.return_value = None
+        mock_sub.return_value = _FULL_SHA
+        assert resolve_git_commit_sha("/some/path") == _FULL_SHA
+        mock_sub.assert_called_once()
+
+
+class TestReadGitCommitShaViaSubprocess:
+    @patch("subprocess.run")
+    def test_read_success(self, mock_run: MagicMock) -> None:
+        mock_run.return_value.returncode = 0
+        mock_run.return_value.stdout = f"{_FULL_SHA}\n"
+        assert read_git_commit_sha_via_subprocess("/some/path") == _FULL_SHA
+
+    @patch("subprocess.run")
+    def test_read_timeout(self, mock_run: MagicMock) -> None:
+        mock_run.side_effect = subprocess.TimeoutExpired(cmd="git", timeout=2)
+        assert read_git_commit_sha_via_subprocess("/some/path") == ""
+
+    @patch("subprocess.run")
+    def test_read_file_not_found(self, mock_run: MagicMock) -> None:
+        mock_run.side_effect = FileNotFoundError()
+        assert read_git_commit_sha_via_subprocess("/some/path") == ""
+
+    @patch("subprocess.run")
+    def test_read_os_error(self, mock_run: MagicMock) -> None:
+        mock_run.side_effect = OSError("error")
+        assert read_git_commit_sha_via_subprocess("/some/path") == ""
+
+    @patch("subprocess.run")
+    def test_read_failure_code(self, mock_run: MagicMock) -> None:
+        mock_run.return_value.returncode = 128
+        assert read_git_commit_sha_via_subprocess("/some/path") == ""
+
+
+class TestReadGitRemoteUrlFromFilesystem:
+    def _write_config(self, tmp_path: Path, body: str) -> None:
+        git_dir = tmp_path / ".git"
+        git_dir.mkdir(exist_ok=True)
+        (git_dir / "config").write_text(body)
+
+    def test_reads_origin_url(self, tmp_path: Path) -> None:
+        self._write_config(
+            tmp_path,
+            '[remote "origin"]\n'
+            "\turl = https://github.com/langchain-ai/deepagents.git\n"
+            "\tfetch = +refs/heads/*:refs/remotes/origin/*\n",
+        )
+        assert (
+            read_git_remote_url_from_filesystem(tmp_path)
+            == "https://github.com/langchain-ai/deepagents.git"
+        )
+
+    def test_ignores_other_remotes(self, tmp_path: Path) -> None:
+        self._write_config(
+            tmp_path,
+            '[remote "upstream"]\n\turl = https://github.com/other/fork.git\n'
+            '[remote "origin"]\n\turl = git@github.com:langchain-ai/deepagents.git\n',
+        )
+        assert (
+            read_git_remote_url_from_filesystem(tmp_path)
+            == "git@github.com:langchain-ai/deepagents.git"
+        )
+
+    def test_no_origin_returns_none(self, tmp_path: Path) -> None:
+        self._write_config(tmp_path, "[core]\n\tbare = false\n")
+        assert read_git_remote_url_from_filesystem(tmp_path) is None
+
+    def test_not_in_repo_returns_empty(self, tmp_path: Path) -> None:
+        assert read_git_remote_url_from_filesystem(tmp_path) == ""
+
+    def test_section_header_matched_case_insensitively(self, tmp_path: Path) -> None:
+        # git treats section names case-insensitively; the reader mirrors that.
+        self._write_config(
+            tmp_path,
+            '[REMOTE "ORIGIN"]\n\turl = https://github.com/org/repo.git\n',
+        )
+        assert (
+            read_git_remote_url_from_filesystem(tmp_path)
+            == "https://github.com/org/repo.git"
+        )
+
+    def test_section_header_extra_whitespace(self, tmp_path: Path) -> None:
+        # Internal spacing inside the header is normalized before matching.
+        self._write_config(
+            tmp_path,
+            '[remote   "origin"]\n\turl = https://github.com/org/repo.git\n',
+        )
+        assert (
+            read_git_remote_url_from_filesystem(tmp_path)
+            == "https://github.com/org/repo.git"
+        )
+
+
+class TestResolveGitRemoteUrl:
+    @patch("deepagents_code._git.read_git_remote_url_via_subprocess")
+    @patch("deepagents_code._git.read_git_remote_url_from_filesystem")
+    def test_empty_string_skips_subprocess(
+        self, mock_fs: MagicMock, mock_sub: MagicMock
+    ) -> None:
+        mock_fs.return_value = ""
+        assert resolve_git_remote_url("/some/path") == ""
+        mock_sub.assert_not_called()
+
+    @patch("deepagents_code._git.read_git_remote_url_via_subprocess")
+    @patch("deepagents_code._git.read_git_remote_url_from_filesystem")
+    def test_fallback_on_none(self, mock_fs: MagicMock, mock_sub: MagicMock) -> None:
+        mock_fs.return_value = None
+        mock_sub.return_value = "https://github.com/org/repo.git"
+        assert resolve_git_remote_url("/p") == "https://github.com/org/repo.git"
+        mock_sub.assert_called_once()
+
+
+class TestReadGitRemoteUrlViaSubprocess:
+    @patch("subprocess.run")
+    def test_read_success(self, mock_run: MagicMock) -> None:
+        mock_run.return_value.returncode = 0
+        mock_run.return_value.stdout = "https://github.com/org/repo.git\n"
+        assert (
+            read_git_remote_url_via_subprocess("/some/path")
+            == "https://github.com/org/repo.git"
+        )
+
+    @patch("subprocess.run")
+    def test_read_timeout(self, mock_run: MagicMock) -> None:
+        mock_run.side_effect = subprocess.TimeoutExpired(cmd="git", timeout=2)
+        assert read_git_remote_url_via_subprocess("/some/path") == ""
+
+    @patch("subprocess.run")
+    def test_read_file_not_found(self, mock_run: MagicMock) -> None:
+        mock_run.side_effect = FileNotFoundError()
+        assert read_git_remote_url_via_subprocess("/some/path") == ""
+
+    @patch("subprocess.run")
+    def test_read_os_error(self, mock_run: MagicMock) -> None:
+        mock_run.side_effect = OSError("error")
+        assert read_git_remote_url_via_subprocess("/some/path") == ""
+
+    @patch("subprocess.run")
+    def test_read_failure_code(self, mock_run: MagicMock) -> None:
+        mock_run.return_value.returncode = 128
+        assert read_git_remote_url_via_subprocess("/some/path") == ""
+
+
+class TestParseRepositoryMetadata:
+    def test_https_github(self) -> None:
+        assert parse_repository_metadata(
+            "https://github.com/langchain-ai/deepagents.git"
+        ) == (
+            "https://github.com/langchain-ai/deepagents",
+            "github",
+            "langchain-ai/deepagents",
+        )
+
+    def test_scp_style_ssh(self) -> None:
+        assert parse_repository_metadata(
+            "git@github.com:langchain-ai/deepagents.git"
+        ) == (
+            "https://github.com/langchain-ai/deepagents",
+            "github",
+            "langchain-ai/deepagents",
+        )
+
+    def test_https_trailing_slash_after_git_suffix(self) -> None:
+        assert parse_repository_metadata("https://github.com/org/repo.git/") == (
+            "https://github.com/org/repo",
+            "github",
+            "org/repo",
+        )
+
+    def test_scp_style_ssh_trailing_slash_after_git_suffix(self) -> None:
+        assert parse_repository_metadata("git@github.com:org/repo.git/") == (
+            "https://github.com/org/repo",
+            "github",
+            "org/repo",
+        )
+
+    def test_strips_embedded_credentials(self) -> None:
+        result = parse_repository_metadata(
+            "https://user:token@gitlab.com/group/project.git"
+        )
+        assert result is not None
+        url, provider, name = result
+        assert url == "https://gitlab.com/group/project"
+        assert provider == "gitlab"
+        assert name == "group/project"
+
+    def test_bitbucket_provider(self) -> None:
+        result = parse_repository_metadata("https://bitbucket.org/team/repo.git")
+        assert result is not None
+        assert result[1] == "bitbucket"
+
+    def test_unknown_host_is_other(self) -> None:
+        result = parse_repository_metadata("git@git.example.com:internal/tool.git")
+        assert result is not None
+        _, provider, name = result
+        assert provider == "other"
+        assert name == "internal/tool"
+
+    def test_nested_group_path(self) -> None:
+        result = parse_repository_metadata(
+            "https://gitlab.com/group/subgroup/project.git"
+        )
+        assert result is not None
+        assert result[2] == "group/subgroup/project"
+
+    def test_ssh_scheme_url(self) -> None:
+        # `ssh://` has a scheme, so it routes through the urlparse branch (not
+        # the scp-style branch) — distinct code path worth pinning.
+        assert parse_repository_metadata(
+            "ssh://git@github.com/langchain-ai/deepagents.git"
+        ) == (
+            "https://github.com/langchain-ai/deepagents",
+            "github",
+            "langchain-ai/deepagents",
+        )
+
+    def test_ssh_scheme_url_with_port(self) -> None:
+        # The port in the authority must not leak into the normalized host/url.
+        assert parse_repository_metadata(
+            "ssh://git@github.com:22/langchain-ai/deepagents.git"
+        ) == (
+            "https://github.com/langchain-ai/deepagents",
+            "github",
+            "langchain-ai/deepagents",
+        )
+
+    def test_returns_named_tuple_fields(self) -> None:
+        # The result exposes named fields, not just positional slots.
+        result = parse_repository_metadata("https://github.com/org/repo.git")
+        assert result is not None
+        assert isinstance(result, RepositoryMetadata)
+        assert result.url == "https://github.com/org/repo"
+        assert result.provider == "github"
+        assert result.name == "org/repo"
+
+    def test_empty_returns_none(self) -> None:
+        assert parse_repository_metadata("") is None
+        assert parse_repository_metadata("   ") is None
+
+    def test_malformed_returns_none(self) -> None:
+        assert parse_repository_metadata("not-a-url") is None

libs/code/tests/unit_tests/test_sessions.py

@@ -531,6 +531,64 @@ class TestTextualSessionState:
         state.reset_thread()
         assert state.approval_mode_key is None
 
+    def test_advance_turn_increments_and_generates_id(self):
+        """advance_turn bumps a 1-based turn_number and yields a fresh turn_id."""
+        state = TextualSessionState(thread_id="t")
+        assert state.turn_number == 0
+        assert state.turn_id is None
+
+        turn_id_1, turn_number_1 = state.advance_turn()
+        assert turn_number_1 == 1
+        assert state.turn_number == 1
+        assert state.turn_id == turn_id_1
+        assert uuid.UUID(turn_id_1)  # valid uuid
+
+        turn_id_2, turn_number_2 = state.advance_turn()
+        assert turn_number_2 == 2
+        assert turn_id_2 != turn_id_1
+
+    def test_reset_thread_resets_turn_markers(self):
+        """reset_thread restarts the per-thread turn sequence."""
+        state = TextualSessionState(thread_id="t")
+        state.advance_turn()
+        state.advance_turn()
+        assert state.turn_number == 2
+
+        state.reset_thread()
+        assert state.turn_number == 0
+        assert state.turn_id is None
+
+    def test_thread_switch_resets_turn_markers(self):
+        """Assigning a different thread_id must not carry the prior turn count.
+
+        `/threads` switches and resume injection set `thread_id` directly
+        (not via `reset_thread`); the per-thread turn sequence has to restart so
+        the switched-to thread's traces aren't ordered under the previous
+        thread's turn_number/turn_id.
+        """
+        state = TextualSessionState(thread_id="thread-a")
+        state.advance_turn()
+        state.advance_turn()
+        assert state.turn_number == 2
+
+        state.thread_id = "thread-b"
+        assert state.turn_number == 0
+        assert state.turn_id is None
+
+        turn_id, turn_number = state.advance_turn()
+        assert turn_number == 1
+        assert state.turn_id == turn_id
+
+    def test_thread_id_reassigned_same_value_keeps_turn_markers(self):
+        """Re-assigning the identical thread_id is a no-op for the turn markers."""
+        state = TextualSessionState(thread_id="thread-a")
+        state.advance_turn()
+        turn_id = state.turn_id
+
+        state.thread_id = "thread-a"
+        assert state.turn_number == 1
+        assert state.turn_id == turn_id
+
 
 class TestFindSimilarThreads:
     """Tests for find_similar_threads function."""

libs/code/tests/unit_tests/test_textual_adapter.py

@@ -660,8 +660,64 @@ class TestBuildStreamConfig:
     """Tests for `build_stream_config` metadata construction."""
 
     def setup_method(self) -> None:
-        """Clear the git-branch cache between tests."""
+        """Clear the git lookup caches between tests."""
         config_module._git_branch_cache.clear()
+        config_module._repo_metadata_cache.clear()
+
+    @pytest.fixture(autouse=True)
+    def _hermetic_git(self) -> Generator[None, None, None]:
+        """Stub git/repo lookups so tests don't read the host repo's real `.git`.
+
+        These tests assert on the identity/turn keys, not on git attribution, so
+        pinning the repo/commit lookups keeps them deterministic in exported
+        checkouts (e.g. a CI tarball with no `.git`).
+        """
+        with (
+            patch.object(config_module, "_get_repository_metadata", return_value=None),
+            patch.object(config_module, "_get_git_commit_sha", return_value=None),
+        ):
+            yield
+
+    def test_coding_agent_identity_block_present(self) -> None:
+        """The coding-agent-v1 identity block is stamped on every config."""
+        from deepagents_code._version import __version__
+
+        metadata = build_stream_config("t-id", assistant_id=None)["metadata"]
+        assert metadata["ls_agent_kind"] == "coding_agent"
+        assert metadata["ls_integration"] == "deepagents-code"
+        assert metadata["ls_agent_runtime"] == "Deep Agents Code"
+        assert metadata["ls_trace_schema_version"] == "coding-agent-v1"
+        assert metadata["ls_integration_version"] == __version__
+        assert metadata["ls_agent_runtime_version"] == __version__
+
+    def test_thread_id_set_as_top_level_metadata(self) -> None:
+        """thread_id is mirrored to top-level metadata for contract grouping."""
+        config = build_stream_config("t-group", assistant_id=None)
+        assert config["metadata"]["thread_id"] == "t-group"
+        assert config["configurable"]["thread_id"] == "t-group"
+
+    def test_turn_markers_passed_through(self) -> None:
+        """turn_id / turn_number reach metadata when provided."""
+        metadata = build_stream_config(
+            "t-turn", assistant_id=None, turn_id="turn-9", turn_number=4
+        )["metadata"]
+        assert metadata["turn_id"] == "turn-9"
+        assert metadata["turn_number"] == 4
+
+    def test_turn_markers_absent_when_unset(self) -> None:
+        """turn_id / turn_number are omitted when not provided."""
+        metadata = build_stream_config("t-noturn", assistant_id=None)["metadata"]
+        assert "turn_id" not in metadata
+        assert "turn_number" not in metadata
+
+    def test_scope_restricted_keys_not_emitted(self) -> None:
+        """approval_policy / ls_subagent_* are never stamped trace-wide."""
+        metadata = build_stream_config(
+            "t-scope", assistant_id="agent", turn_id="t", turn_number=1
+        )["metadata"]
+        assert "approval_policy" not in metadata
+        assert "ls_subagent_id" not in metadata
+        assert "ls_subagent_type" not in metadata
 
     def test_dcode_agent_fields_present(self) -> None:
         """Selected dcode agent metadata should be present."""
@@ -825,6 +881,27 @@ class TestGetGitBranch:
         mock_resolve.assert_called_once_with("/tmp/repo")
 
 
+class TestGetGitCommitSha:
+    """Tests for `_get_git_commit_sha` freshness."""
+
+    def test_resolves_commit_fresh_on_each_call(self) -> None:
+        """HEAD moves mid-session, so the SHA must be re-resolved every call."""
+        with (
+            patch(
+                "deepagents_code.config.Path.cwd",
+                return_value=Path("/tmp/repo"),
+            ),
+            patch(
+                "deepagents_code._git.resolve_git_commit_sha",
+                side_effect=["sha-before", "sha-after"],
+            ) as mock_resolve,
+        ):
+            assert config_module._get_git_commit_sha() == "sha-before"
+            assert config_module._get_git_commit_sha() == "sha-after"
+
+        assert mock_resolve.call_count == 2
+
+
 class TestGetGitBranchOSError:
     """Tests for _get_git_branch when Path.cwd() raises OSError."""
 
@@ -845,8 +922,9 @@ class TestBuildStreamConfigOSError:
     """Tests for build_stream_config when Path.cwd() raises OSError."""
 
     def setup_method(self) -> None:
-        """Clear the git-branch cache between tests."""
+        """Clear the git lookup caches between tests."""
         config_module._git_branch_cache.clear()
+        config_module._repo_metadata_cache.clear()
 
     def test_cwd_absent_on_oserror(self) -> None:
         """Cwd should be absent from metadata when Path.cwd() raises."""
@@ -1048,6 +1126,7 @@ class _SequencedAgent:
         self._streams_by_call = streams_by_call
         self.stream_inputs: list[dict | Command] = []
         self.contexts: list[Any] = []
+        self.configs: list[Any] = []
         self.store_items: list[tuple[tuple[str, ...], str, dict[str, Any]]] = []
 
     async def aput_store_item(
@@ -1064,6 +1143,7 @@ class _SequencedAgent:
         stream_input: dict | Command,
         *_: Any,
         context: object = None,
+        config: object = None,
         **__: Any,
     ) -> AsyncIterator[tuple[Any, ...]]:
         """Yield chunks for this invocation and record stream inputs/context.
@@ -1075,6 +1155,7 @@ class _SequencedAgent:
         """
         self.stream_inputs.append(stream_input)
         self.contexts.append(dict(context) if isinstance(context, dict) else context)
+        self.configs.append(config)
         chunks = self._streams_by_call.pop(0) if self._streams_by_call else []
         for chunk in chunks:
             yield chunk
@@ -1095,6 +1176,58 @@ class _FailingApprovalStoreAgent(_SequencedAgent):
         raise RuntimeError(msg)
 
 
+class TestExecuteTaskTextualTurnMarkers:
+    """End-to-end: turn markers advance and reach the stream config metadata."""
+
+    async def test_turn_markers_flow_into_stream_config_and_advance(self) -> None:
+        """A real session state advances turn markers into each turn's config.
+
+        Guards the full wiring (`advance_turn` -> `build_stream_config` ->
+        `astream` config) that the per-piece unit tests don't exercise together:
+        a dropped `advance_turn()` call or mis-passed turn tuple would still pass
+        those, but not this.
+        """
+        from deepagents_code.app import TextualSessionState
+
+        session_state = TextualSessionState(thread_id="thread-1", auto_approve=True)
+        agent = _SequencedAgent([[], []])
+        adapter = TextualUIAdapter(
+            mount_message=_mock_mount,
+            update_status=_noop_status,
+            request_approval=_mock_approval,
+        )
+
+        # Stub git lookups so the captured metadata is deterministic.
+        with (
+            patch.object(config_module, "_get_repository_metadata", return_value=None),
+            patch.object(config_module, "_get_git_commit_sha", return_value=None),
+        ):
+            await execute_task_textual(
+                user_input="first",
+                agent=agent,
+                assistant_id="assistant",
+                session_state=session_state,
+                adapter=adapter,
+            )
+            await execute_task_textual(
+                user_input="second",
+                agent=agent,
+                assistant_id="assistant",
+                session_state=session_state,
+                adapter=adapter,
+            )
+
+        first_meta = agent.configs[0]["metadata"]
+        second_meta = agent.configs[1]["metadata"]
+        assert first_meta["turn_number"] == 1
+        assert second_meta["turn_number"] == 2
+        assert first_meta["turn_id"]
+        assert second_meta["turn_id"]
+        assert first_meta["turn_id"] != second_meta["turn_id"]
+        # The session state itself reflects the latest turn.
+        assert session_state.turn_number == 2
+
+
 class TestExecuteTaskTextualAutoApproveInput:
     """Auto-approve must ride on run context, never a first-turn `Command`."""