Investigating Agentforce Architecture

name: describe_sobject
argv:
  - sf
  - sobject
  - describe
  - --sobject
  - "{{SOBJECT}}"
  - --target-org
  - "{{ORG_ALIAS}}"
  - --json
timeout_seconds: 300
required_params: [ORG_ALIAS, SOBJECT]
success_check: stdout_json_status_zero
auth_required_stderr_patterns:
  - NoOrgAuthenticationError
  - AuthInfoError

name: describe_tooling_sobject
argv:
  - sf
  - sobject
  - describe
  - --sobject
  - "{{SOBJECT}}"
  - --use-tooling-api
  - --target-org
  - "{{ORG_ALIAS}}"
  - --json
timeout_seconds: 300
required_params: [ORG_ALIAS, SOBJECT]
success_check: stdout_json_status_zero
auth_required_stderr_patterns:
  - NoOrgAuthenticationError
  - AuthInfoError

name: list_metadata_genaiprompttemplate
argv:
  - sf
  - org
  - list
  - metadata
  - --metadata-type
  - GenAiPromptTemplate
  - --target-org
  - "{{ORG_ALIAS}}"
  - --json
timeout_seconds: 300
required_params: [ORG_ALIAS]
success_check: stdout_json_status_zero
auth_required_stderr_patterns:
  - NoOrgAuthenticationError
  - AuthInfoError

name: org_display
argv:
  - sf
  - org
  - display
  - --target-org
  - "{{ORG_ALIAS}}"
  - --json
  - --verbose
timeout_seconds: 300
required_params: [ORG_ALIAS]
success_check: stdout_json_status_zero
auth_required_stderr_patterns:
  - NoOrgAuthenticationError
  - AuthInfoError

name: retrieve_genai_plugin
argv:
  - sf
  - project
  - retrieve
  - start
  - --target-org
  - "{{ORG_ALIAS}}"
  - --target-metadata-dir
  - "{{TARGET_DIR}}"
  - --json
extra_argv_anchor: --target-metadata-dir
timeout_seconds: 300
required_params: [ORG_ALIAS, TARGET_DIR]
success_check: stdout_json_status_zero
auth_required_stderr_patterns:
  - NoOrgAuthenticationError
  - AuthInfoError

name: show_access_token
# Primary access-token retrieval path per forcedotcom/cli#3560 (effective
# 2026-05-27). The `sf org display --json` field is now redacted to a
# placeholder string by default; the dedicated `sf org auth show-access-token`
# command is the long-term replacement.
#
# `--no-prompt` skips the interactive confirmation banner so the command
# behaves identically under `--json`.
#
# The legacy fallback (`sf org display` with `SF_TEMP_SHOW_SECRETS=true`)
# stays wired for sf CLI versions that don't ship show-access-token; that
# fallback is orchestrated in `main.py::_resolve_creds`, not here.
argv:
  - sf
  - org
  - auth
  - show-access-token
  - --target-org
  - "{{ORG_ALIAS}}"
  - --json
  - --no-prompt
timeout_seconds: 60
required_params: [ORG_ALIAS]
success_check: stdout_json_status_zero
auth_required_stderr_patterns:
  - NoOrgAuthenticationError
  - AuthInfoError

%% action_tree.mmd - section 4 declared action tree (flowchart TB)
%%
%% Placeholders (substituted by render_architecture.load_mermaid):
%%   SUBGRAPHS placeholder: one `subgraph <topic_id>[<topic_label>] ... end`
%%                          block per topic. Non-topic actions (plannerActions)
%%                          go in a synthetic subgraph keyed `_plannerActions`.
%%   EDGES placeholder:     tree edges, one per child->parent descent. Cycle-
%%                          back annotations render as dotted back-edges
%%                          `A -.->|cycle| B` (see
%%                          render_architecture._render_action_tree).
%%
%% placeholder names are written as bare tokens (no double-curly)
%% so load_mermaid's single-pass str.replace cannot corrupt this header.
%%
%% Node-cap enforcement: if the tree exceeds `max_mermaid_nodes["flowchart"]`,
%% the renderer emits a summary placeholder instead of this template.
flowchart TB
    {{SUBGRAPHS}}

    {{EDGES}}

%% data_flow.mmd - section 8 data flow / context propagation (flowchart LR)
%%
%% Placeholders (substituted by render_architecture.load_mermaid):
%%   NODES placeholder: node declarations: `<id>[<label>]`, one per line.
%%   EDGES placeholder: labeled-where-possible edges:
%%                      `A -->|<var>: <type>| B`, bare `A --> B` when no
%%                      parameter threads between the lanes.
%%
%% placeholder names are written as bare tokens (no double-curly)
%% so load_mermaid's single-pass str.replace cannot corrupt this header.
%%
%% Source: GenAiPlannerAttrDefinition rows on the planner drive which slots
%% populate which actions. Renderer (render_architecture._render_data_flow)
%% builds nodes/edges from the tree's planner-slot metadata; no slots ->
%% degenerate diagram with just `User --> Planner --> Actions` skeleton.
flowchart LR
    {{NODES}}

    {{EDGES}}

%% dependency_graph.mmd - conditional section (graph LR)
%%
%% Rendered ONLY when `_unresolved[]` is non-empty OR cycles exist in the
%% action tree. Otherwise render_architecture skips this template entirely.
%%
%% Placeholders (substituted by render_architecture.load_mermaid):
%%   NODES placeholder: node declarations: `<id>[<label>]`. Unresolved refs
%%                      use a `:::unresolved` class marker for CSS styling.
%%   EDGES placeholder: cross-artifact dependency edges harvested from the
%%                      tree. Cycle-back edges render as `A -.->|cycle| B`.
%%
%% placeholder names are written as bare tokens (no double-curly)
%% so load_mermaid's single-pass str.replace cannot corrupt this header.
graph LR
    {{NODES}}

    {{EDGES}}

    classDef unresolved stroke-dasharray: 5 5

%% invocation_sequence.mmd - section 3 invocation sequence (sequenceDiagram)
%%
%% Placeholders (substituted by render_architecture.load_mermaid):
%%   PARTICIPANTS placeholder: one `participant <Name>` line per lane,
%%                             newline-joined; deactivate with
%%                             `<Dst>-->>-<Src>: <response>`.
%%   MESSAGES placeholder:     one `<Src>->>+<Dst>: <label>` per message line.
%%
%% placeholder names above are written as bare tokens (no double-
%% curly) so load_mermaid's single-pass str.replace cannot substitute them
%% here. That keeps %% headers intact in the rendered output - Mermaid
%% ignores %% lines as comments at render time, and debuggability wins.
%%
%% Renderer (render_architecture.py::_render_invocation_sequence) is the sole
%% consumer. Generation-aware: classic ReAct uses Planner<->TopicClassifier
%% <->ActionExecutor; NGA orchestrations add one lane per sub-agent.
sequenceDiagram
    autonumber
    {{PARTICIPANTS}}
    {{MESSAGES}}

%% planner_state.mmd - section 7 planner state machine (stateDiagram-v2)
%%
%% Placeholders (substituted by render_architecture.load_mermaid):
%%   STATES placeholder:      state declarations (`state Foo { ... }` or
%%                            bare `state`).
%%   TRANSITIONS placeholder: transition list (`A --> B: <label>`).
%%
%% placeholder names are written as bare tokens (no double-curly)
%% so load_mermaid's single-pass str.replace cannot corrupt this header.
%%
%% Renderer selects one of a small library of per-generation state machines
%% (render_architecture._planner_state_for_generation) and fills in the
%% template. BYOP / SEARCH generations skip this section and emit a prose
%% placeholder at call site rather than rendering this template.
stateDiagram-v2
    {{STATES}}

    {{TRANSITIONS}}

SELECT Id, Name, Body, SymbolTable, ApiVersion, IsValid
FROM ApexClass
WHERE Id IN ({{APEX_IDS_LIST}})

SELECT Id, Name, Body, SymbolTable, ApiVersion, IsValid
FROM ApexClass
WHERE Name IN ({{NAMES_LIST}})

SELECT DeveloperName, MasterLabel, Description, AgentType, Type, AgentTemplate, BotSource
FROM BotDefinition
WHERE DeveloperName = '{{AGENT_API_NAME}}'

SELECT Id, DeveloperName, Status, BotDefinitionId,
       BotDefinition.DeveloperName, BotDefinition.MasterLabel
FROM BotVersion
WHERE BotDefinition.DeveloperName = '{{AGENT_API_NAME}}'

SELECT Id, DeveloperName, ActiveVersionId, LatestVersionId
FROM FlowDefinition
WHERE Id IN ({{FLOW_DEF_IDS_LIST}})

SELECT Id, DeveloperName, NamespacePrefix, ActiveVersionId
FROM FlowDefinition
WHERE DeveloperName IN ({{NAMES_LIST}}) AND NamespacePrefix = NULL

SELECT DurableId, ApiName, Label, NamespacePrefix, ActiveVersionId,
       IsActive, ManageableState, ProcessType
FROM FlowDefinitionView
WHERE DurableId IN ({{DURABLE_IDS_LIST}})

SELECT Id, FullName, Metadata
FROM Flow
WHERE Id = '{{FLOW_VERSION_ID}}'

SELECT Id, DeveloperName, MasterLabel, Description, InvocationTargetType, InvocationTarget,
       IsLocal, IsConfirmationRequired, IsIncludeInProgressIndicator, ProgressIndicatorMessage,
       Source, PluginId, PlannerId, ParentId, LocalDeveloperName
FROM GenAiFunctionDefinition
WHERE PluginId IN ({{PLUGIN_IDS}})

SELECT Id, ParentId, DeveloperName, MasterLabel, Description, MappingType, ParameterName
FROM GenAiPlannerAttrDefinition
WHERE ParentId IN ({{PARENT_IDS}})

SELECT Id, PlannerId, Plugin
FROM GenAiPlannerFunctionDef
WHERE PlannerId = '{{PLANNER_ID}}'

SELECT Id, DeveloperName, MasterLabel, Description, PlannerType, Capabilities, AgentGraph
FROM GenAiPlannerDefinition
WHERE DeveloperName LIKE '{{AGENT_NAME}}%\_{{VERSION}}'

SELECT Id, PluginId, Function
FROM GenAiPluginFunctionDef
WHERE PluginId IN ({{PLUGIN_IDS}})

SELECT Id, GenAiPluginDefinitionId, DeveloperName, MasterLabel, Description, SortOrder
FROM GenAiPluginInstructionDef
WHERE GenAiPluginDefinitionId IN ({{PLUGIN_IDS}})

SELECT Id, DeveloperName, MasterLabel, Description, PluginType, Scope,
       IsLocal, CanEscalate, Source, ParentId, LocalDeveloperName
FROM GenAiPluginDefinition
WHERE PlannerId = '{{PLANNER_ID}}'

{
  "_doc": "retired-name references replaced with plugin-scoped layout. Machine-readable I/O contract for investigating-agentforce-architecture. Orchestrator code reads this file (not SKILL.md prose) to discover the input schema, output keys, STATUS enum, and error-contract rules. The architecture skill runs inline (no subagent).",
  "_schema_version": "3.1",
  "_schema_notes": "3.1 (2026-05-05) canonicalizes `invocation_type` on STANDARD_ACTION / UNKNOWN nodes in the metadata tree. Prior versions split this across `raw_invocation_type` (bundle-sourced nodes) and `raw_action_type` (flow-actionCall-sourced nodes). Both legacy keys are still tolerated by readers for one release.",
  "input": {
    "required": {
      "org_alias": {
        "type": "string",
        "description": "sf CLI alias (e.g. 'my-org-alias'). Auth is always derived via `sf org display --target-org $ORG_ALIAS --json`."
      },
      "agent_api_name": {
        "type": "string",
        "regex": "^[A-Za-z0-9_]+$",
        "description": "BotDefinition.DeveloperName (e.g. 'MyAgent'). Must match the regex above; INVALID_INPUT otherwise."
      }
    },
    "optional": {
      "agent_version_api_name": {
        "type": "string",
        "description": "BotVersion.DeveloperName (e.g. 'v5'). When omitted, the agent auto-picks the Active version with the highest natural-key sort (v10 > v9); VERSION_AUTO_PICKED=true in the RESULT block."
      },
      "org_id_15": {
        "type": "string",
        "regex": "^[A-Za-z0-9]{15}$",
        "description": "15-char org ID. If omitted, derived from `sf org display`."
      },
      "org_id_18": {
        "type": "string",
        "regex": "^[A-Za-z0-9]{18}$",
        "description": "18-char org ID. If omitted, derived from `sf org display`."
      },
      "session_id": {
        "type": "string",
        "description": "Only used to name the ephemeral $WORK_DIR. Random UUID if omitted."
      },
      "work_dir": {
        "type": "absolute_path",
        "description": "Ephemeral scratch. Default: /tmp/investigating-agentforce-architecture-<epoch>-<rand>."
      },
      "cache_root": {
        "type": "absolute_path",
        "description": "Internal rebuildable cache root. Default: ~/.vibe/cache/investigating-agentforce-architecture. Override with --cache-dir <path>."
      },
      "data_root": {
        "type": "absolute_path",
        "description": "Durable user-facing output root. Default: ~/.vibe/data/investigating-agentforce-architecture. Override with --data-dir <path>."
      },
      "force_refresh": {
        "type": "boolean",
        "description": "If true, ignore cache and rewrite. Default: false."
      }
    },
    "minimum_viable": {
      "org_alias": "my-org-alias",
      "agent_api_name": "MyAgent"
    },
    "notes": [
      "Standalone and orchestrated modes accept the same input fields; no renames.",
      "To bypass cache, delete the cache entry or pass force_refresh: true."
    ]
  },
  "output": {
    "format": "Two-part return message: (1) one-line prose status, (2) blank line, (3) === RESULT === key-value block. The KV block is always LAST in the output and always present.",
    "parse_rule": "Find the line '=== RESULT ===', then read subsequent lines as KEY=VALUE pairs until EOF. Values are raw strings; no quoting; one pair per line.",
    "parse_examples": {
      "bash": "awk -F= '/^STATUS=/{print $2; exit}' <<< \"$subagent_output\"",
      "python": "import re; dict(line.split('=', 1) for line in output.splitlines() if re.match(r'^[A-Z_]+=', line))"
    },
    "keys": {
      "STATUS": {
        "type": "enum",
        "presence": "always",
        "description": "Terminal status. See error_contract for per-status required keys.",
        "values": [
          "OK",
          "PARTIAL_OK",
          "INVALID_INPUT",
          "AUTH_REQUIRED",
          "AGENT_NOT_FOUND",
          "AGENT_VERSION_NOT_FOUND",
          "RETRIEVE_FAILED",
          "WRITE_FAILED"
        ]
      },
      "ERROR_DETAIL": {
        "type": "string",
        "presence": "on_error",
        "description": "Human-readable explanation on any non-OK STATUS. Sanitized (no backticks/quotes/dollar signs/newlines)."
      },
      "AGENT_API_NAME": {
        "type": "string",
        "presence": "always",
        "description": "Echo of input agent_api_name. May be empty on INVALID_INPUT if sanitize rejected before resolution."
      },
      "AGENT_VERSION": {
        "type": "string",
        "presence": "on_success_or_after_resolve",
        "description": "Resolved BotVersion.DeveloperName (e.g. 'v5')."
      },
      "VERSION_AUTO_PICKED": {
        "type": "boolean",
        "presence": "on_success_or_after_resolve",
        "description": "True iff agent_version_api_name was omitted and the agent auto-picked."
      },
      "AGENT_GENERATION": {
        "type": "enum",
        "presence": "always",
        "values": ["classic", "nga", "unknown"],
        "description": "classic = AiCopilot__ planner, nga = Atlas__ planner, unknown = anything else / not resolved."
      },
      "BOT_ID": {
        "type": "string",
        "presence": "on_success_or_after_resolve",
        "description": "18-char BotDefinition.Id from SOQL."
      },
      "ORG_ID_15": {
        "type": "string",
        "presence": "on_success_or_after_auth",
        "description": "15-char org ID. Cache-key prefix."
      },
      "ORG_ID_18": {
        "type": "string",
        "presence": "on_success_or_after_auth",
        "description": "18-char org ID (from `sf org display`)."
      },
      "OUTPUT_JSON_PATH": {
        "type": "absolute_path",
        "presence": "on_success",
        "description": "Durable {agent}_{version}_metadata_tree.json path under the resolved data_root."
      },
      "OUTPUT_SUMMARY_PATH": {
        "type": "absolute_path",
        "presence": "always_empty",
        "description": "Reserved — historically pointed at {agent}_{version}_metadata_tree.summary.md. Dropped in ; field is now always emitted empty for RESULT-block shape stability."
      },
      "CACHE_PATH": {
        "type": "absolute_path",
        "presence": "on_success_or_after_cache_check",
        "description": "Cache dir ($CACHE_ROOT/$ORG_ID_15/$AGENT__$VERSION/). Ends with /."
      },
      "CACHE_HIT": {
        "type": "boolean",
        "presence": "on_success_or_after_cache_check",
        "description": "True iff the cache was served without any Metadata API retrieves."
      },
      "CACHED_AT_UTC": {
        "type": "iso8601_utc_or_empty",
        "presence": "on_success_or_after_cache_check",
        "description": "ISO-8601 UTC built-at time. On cache hit this is the age of the cached snapshot; on cold build this is the just-computed timestamp."
      },
      "NODE_COUNT": {
        "type": "integer",
        "presence": "on_success",
        "description": "Total nodes in the declared_action_tree."
      },
      "DEPTH": {
        "type": "integer",
        "presence": "on_success",
        "description": "Max depth of the tree (BOT_DEFINITION = depth 0)."
      },
      "PARTIAL": {
        "type": "boolean",
        "presence": "on_success",
        "description": "True iff the tree is incomplete (planner unresolved OR MAX_WAVE hit). When true, STATUS=PARTIAL_OK."
      },
      "UNRESOLVED_COUNT": {
        "type": "integer",
        "presence": "on_success",
        "description": "Length of tree._unresolved[]."
      },
      "AVAILABLE_BOTS": {
        "type": "csv",
        "presence": "on_AGENT_NOT_FOUND",
        "description": "Comma-separated list of all BotDefinition.DeveloperName values in the org."
      },
      "AVAILABLE_VERSIONS": {
        "type": "csv",
        "presence": "on_AGENT_VERSION_NOT_FOUND",
        "description": "Format: 'v5(Active),v4(Inactive),...' — natural-key sorted DESC."
      },
      "RESULT_BLOCK_PATH": {
        "type": "absolute_path",
        "presence": "always",
        "description": "Path to $DATA_DIR/last_result_block.txt — byte-for-byte tee written BEFORE stdout."
      },
      "WALL_TIME_SECONDS": {
        "type": "float",
        "presence": "always",
        "description": "End-to-end wall-clock seconds. Two decimals. Target: 90-220s cold, <2s warm."
      }
    }
  },
  "error_contract": {
    "OK": {
      "description": "Tree built, zero unresolved, planner resolved.",
      "required_keys": ["STATUS", "AGENT_API_NAME", "AGENT_VERSION", "VERSION_AUTO_PICKED", "AGENT_GENERATION", "BOT_ID", "ORG_ID_15", "ORG_ID_18", "OUTPUT_JSON_PATH", "OUTPUT_SUMMARY_PATH", "CACHE_PATH", "CACHE_HIT", "CACHED_AT_UTC", "NODE_COUNT", "DEPTH", "PARTIAL", "UNRESOLVED_COUNT", "RESULT_BLOCK_PATH", "WALL_TIME_SECONDS"],
      "optional_keys": []
    },
    "PARTIAL_OK": {
      "description": "Tree built but planner unresolved (Bot.bot had no planner) OR waves hit MAX_WAVE=5. tree._unresolved[] explains.",
      "required_keys": ["STATUS", "AGENT_API_NAME", "AGENT_VERSION", "AGENT_GENERATION", "BOT_ID", "ORG_ID_15", "ORG_ID_18", "OUTPUT_JSON_PATH", "CACHE_PATH", "PARTIAL", "UNRESOLVED_COUNT", "WALL_TIME_SECONDS", "RESULT_BLOCK_PATH"],
      "optional_keys": ["OUTPUT_SUMMARY_PATH", "CACHE_HIT", "CACHED_AT_UTC", "NODE_COUNT", "DEPTH", "VERSION_AUTO_PICKED", "ERROR_DETAIL"]
    },
    "INVALID_INPUT": {
      "description": "agent_api_name failed [A-Za-z0-9_]+ regex, or other input validation (org_alias empty, ORG_ID_15 regex, $WORK_DIR symlink/foreign-owned).",
      "required_keys": ["STATUS", "ERROR_DETAIL", "AGENT_API_NAME"],
      "optional_keys": ["ORG_ID_15", "ORG_ID_18"]
    },
    "AUTH_REQUIRED": {
      "description": "`sf org display` returned nothing, or a SOQL/Metadata call failed with an auth error. Caller must run `sf org login web --alias $ORG_ALIAS`.",
      "required_keys": ["STATUS", "ERROR_DETAIL", "AGENT_API_NAME"],
      "optional_keys": ["ORG_ID_15", "ORG_ID_18"]
    },
    "AGENT_NOT_FOUND": {
      "description": "BotDefinition.DeveloperName not in the org. AVAILABLE_BOTS carries the CSV of what IS in the org.",
      "required_keys": ["STATUS", "ERROR_DETAIL", "AGENT_API_NAME", "AVAILABLE_BOTS"],
      "optional_keys": ["ORG_ID_15", "ORG_ID_18"]
    },
    "AGENT_VERSION_NOT_FOUND": {
      "description": "No matching BotVersion under the bot (explicit version doesn't exist, or no Active version found during auto-pick).",
      "required_keys": ["STATUS", "ERROR_DETAIL", "AGENT_API_NAME", "AVAILABLE_VERSIONS"],
      "optional_keys": ["BOT_ID", "ORG_ID_15", "ORG_ID_18"]
    },
    "RETRIEVE_FAILED": {
      "description": "`sf project retrieve start` failed (network, permissions, zip unreadable). ERROR_DETAIL carries the specifics.",
      "required_keys": ["STATUS", "ERROR_DETAIL", "AGENT_API_NAME", "AGENT_VERSION", "BOT_ID"],
      "optional_keys": ["ORG_ID_15", "ORG_ID_18", "OUTPUT_JSON_PATH"]
    },
    "WRITE_FAILED": {
      "description": "Filesystem write failed during finalize (DATA_DIR rename, CACHE_DIR rename, or tree rewrite).",
      "required_keys": ["STATUS", "ERROR_DETAIL", "AGENT_API_NAME", "AGENT_VERSION"],
      "optional_keys": ["BOT_ID", "ORG_ID_15", "ORG_ID_18", "CACHE_HIT", "CACHED_AT_UTC"]
    }
  },
  "stability": {
    "key_names": "Stable. Key-name changes require a NOTICE.md entry and a skill reinstall; consumers read this contract to discover the current key set.",
    "error_codes": "The STATUS enum is fixed at 8 values. A new status is a breaking change for consumers that switch on STATUS.",
    "path_format": "All path values are absolute (no tilde expansion required). One KEY=VALUE per line. No trailing whitespace.",
    "order": "Keys inside the RESULT block appear in a stable order — orchestrators should not depend on order but may use it for human readability."
  },
  "referenced_files": {
    "skill_md": "<SKILL_ROOT>/SKILL.md"
  }
}

# Shared path / fs-guard / sql helpers for investigating-agentforce-architecture.

#!/usr/bin/env python3
r"""Filesystem-safety and input-validation guard.

One script, six check types. Emits a STATUS=INVALID_INPUT RESULT block on
failure + tees to $ERROR_TEE on disk + exits 1.

Guiding principle: the agent MUST suffix every call with `|| exit 1` — Python
`sys.exit(1)` only terminates this subprocess, not the parent bash. A bare
call without `|| exit 1` silently continues past a failed guard, which is
the worst possible failure mode for a security check.

Check types:
    symlink     — path must NOT be a symlink (rejects pre-planted attacker bait)
    owned       — path must be owned by current UID (rejects foreign-owned dirs)
    uuid          — value must match ^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$
    org_id_15     — value must match ^[A-Za-z0-9]{15}$ (Salesforce org ID slice)
    api_name      — value must match ^[A-Za-z0-9_]+$ (Salesforce API identifier)
    api_version   — value must match ^v[0-9]+\.[0-9]+$ (e.g. v60.0)
    agent_version — value must match ^v[0-9]+$ (e.g. v5 — Agentforce agent version, no dot-minor)
    not_empty     — value must be non-empty string

Python-importable API:
    validate_api_name(value, label="...")      — raises ValidationError on bad input
    validate_api_version(value, label="...")   — raises ValidationError on bad input
    validate_agent_version(value, label="...") — raises ValidationError on bad input
    validate_org_id_15(value, label="...")     — raises ValidationError on bad input
    ValidationError                            — carries (label, reason)

The Python API is used by in-process callers (SOQL loader, path builders in
config.py) that need a raise-based boundary rather than the process-exit
CLI behavior. Regex is shared — do NOT duplicate API_NAME_RE / API_VERSION_RE.

Usage:
    python3 fs_guard.py <value> <label> <check> || exit 1

    # Input validation
    python3 fs_guard.py "$AGENT_API_NAME" agent_api_name api_name || exit 1
    python3 fs_guard.py "$ORG_ALIAS"      org_alias      not_empty || exit 1

    # Filesystem safety
    python3 fs_guard.py "$WORK_DIR"  WORK_DIR  symlink || exit 1
    python3 fs_guard.py "$WORK_DIR"  WORK_DIR  owned   || exit 1
    python3 fs_guard.py "$ORG_ID_15" ORG_ID_15 org_id_15 || exit 1

Inputs:
    argv[1]     value or path to check
    argv[2]     label (appears in ERROR_DETAIL; used to identify which guard tripped)
    argv[3]     check type (one of the 6 above)
    env         $ERROR_TEE (optional): path to disk-tee file. Defaults to
                $HOME/.vibe/data/investigating-agentforce-architecture/_last_error_result.txt.
                Also reads $AGENT_API_NAME / $ORG_ID_18 / $ORG_ID_15 for
                RESULT-block context if set.

Outputs:
    on failure: STATUS=INVALID_INPUT RESULT block on stdout + tee to disk, exit 1
    on success: silent, exit 0
    on bad argv: exit 1 with a minimal ERROR_DETAIL ("fs_guard internal: ...")
"""
import os
import pathlib
import re
import sys

UUID_RE = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$")  # @rule-suppress starter-sec-002 — re.compile, not eval/exec
ORG_ID_15_RE = re.compile(r"^[A-Za-z0-9]{15}$")  # @rule-suppress starter-sec-002 — re.compile, not eval/exec
API_NAME_RE = re.compile(r"^[A-Za-z0-9_]+$")  # @rule-suppress starter-sec-002 — re.compile, not eval/exec
# api_version check — matches `v60.0`, `v66.0`, etc. Used before any
# Path composition that embeds the api_version segment.
API_VERSION_RE = re.compile(r"^v[0-9]+\.[0-9]+$")  # @rule-suppress starter-sec-002 — re.compile, not eval/exec
# Agent-version check — matches Agentforce version identifiers (`v1`, `v5`,
# `v12`). Deliberately stricter than api_name (rejects free-form names like
# `release_1`) and looser than api_version (rejects `v66.0` which is a REST
# API version, never an agent version). Lives on its own so callers that
# embed agent_version in a filesystem path get the tight regex, not the
# permissive api_name fallback.
AGENT_VERSION_RE = re.compile(r"^v[0-9]+$")  # @rule-suppress starter-sec-002 — re.compile, not eval/exec
VALID_CHECKS = {"symlink", "owned", "uuid", "org_id_15", "api_name", "api_version", "agent_version", "not_empty"}


# -----------------------------------------------------------------------------
# Python-importable API 
# -----------------------------------------------------------------------------
# In-process callers (soql_loader.load_soql, config.build_*_dir) need a
# raise-based validation boundary, distinct from the CLI script's exit-1
# behavior. The regexes are shared with the CLI checks above — do NOT
# duplicate. If a regex changes, both surfaces update in lockstep.


class ValidationError(ValueError):
    """Raised by the Python-importable validators on bad input.

    Carries the label (which field failed) + reason. Callers (e.g.
    soql_loader.load_soql) convert this into `_unresolved[]` entries or
    RESULT-level INVALID_INPUT responses as appropriate.
    """

    def __init__(self, label: str, reason: str) -> None:
        self.label = label
        self.reason = reason
        super().__init__(f"{label}: {reason}")


def validate_api_name(value, label: str = "value") -> None:
    """Validate `value` against ^[A-Za-z0-9_]+$ (Salesforce API identifier).

    used by soql_loader.load_soql at the substitution boundary to catch
    injection attempts in names pulled from Bot XML, Flow.Metadata, ApexClass
    SymbolTable, etc.

    used by config.build_*_dir helpers for path components that must
    not contain `..`, `/`, or other traversal characters.

    Raises ValidationError on any of: None, non-string, empty, regex miss.
    """
    if value is None:
        raise ValidationError(label, "is None")
    if not isinstance(value, str):
        raise ValidationError(label, f"must be str, got {type(value).__name__}")
    if not value:
        raise ValidationError(label, "must not be empty")
    if not API_NAME_RE.match(value):
        # Preview first 20 chars so logs/_unresolved entries carry enough
        # context to debug, without dumping an unbounded attacker-controlled
        # string into output.
        preview = value[:20]
        raise ValidationError(
            label,
            f"does not match [A-Za-z0-9_]+ (preview={preview!r})",
        )


def validate_api_version(value, label: str = "api_version") -> None:
    """Validate `value` against ^v[0-9]+\\.[0-9]+$ (Salesforce API version).

    used by config.build_*_dir helpers. api_version is returned by
    `sf org display` and later embedded in cache paths; reject anything that
    could escape the cache subtree.
    """
    if value is None:
        raise ValidationError(label, "is None")
    if not isinstance(value, str):
        raise ValidationError(label, f"must be str, got {type(value).__name__}")
    if not value:
        raise ValidationError(label, "must not be empty")
    if not API_VERSION_RE.match(value):
        preview = value[:20]
        raise ValidationError(
            label,
            f"does not match v<major>.<minor> (preview={preview!r})",
        )


def validate_agent_version(value, label: str = "agent_version") -> None:
    """Validate `value` against ^v[0-9]+$ (Agentforce agent version).

    Strictly `v<digits>` with no dot-minor — matches the shape Agentforce
    actually uses (`v1`, `v5`, `v12`). Rejects `release_1`, `v66.0`, `FOO`,
    and anything else that could silently slip past a permissive api_name
    check and land in a filesystem path.

    used by path-builder helpers for the agent_version segment.
    """
    if value is None:
        raise ValidationError(label, "is None")
    if not isinstance(value, str):
        raise ValidationError(label, f"must be str, got {type(value).__name__}")
    if not value:
        raise ValidationError(label, "must not be empty")
    if not AGENT_VERSION_RE.match(value):
        preview = value[:20]
        raise ValidationError(
            label,
            f"does not match v<digits> (preview={preview!r})",
        )


def validate_org_id_15(value, label: str = "org_id_15") -> None:
    """Validate `value` against ^[A-Za-z0-9]{15}$ (Salesforce 15-char org ID).

    stricter than validate_api_name — enforces exact 15 chars AND
    no underscores. `org_id_15` always comes from a Salesforce-generated
    field, never from free-form input.
    """
    if value is None:
        raise ValidationError(label, "is None")
    if not isinstance(value, str):
        raise ValidationError(label, f"must be str, got {type(value).__name__}")
    if not value:
        raise ValidationError(label, "must not be empty")
    if not ORG_ID_15_RE.match(value):
        preview = value[:20]
        raise ValidationError(
            label,
            f"must be exactly 15 alphanumeric chars (preview={preview!r})",
        )


def scrub(s: str) -> str:
    # Same rules as sanitize.py; duplicated locally so this script has no
    # intra-skill imports. Keeps the agent-script boundary clean.
    bad = set("`$\"\\\r\t\0\n")
    return "".join(c for c in (s or "") if c not in bad)


def emit_failure(reason: str, label: str) -> None:
    agent_api_name = scrub(os.environ.get("AGENT_API_NAME", ""))
    org_id_18 = scrub(os.environ.get("ORG_ID_18", ""))
    org_id_15 = scrub(os.environ.get("ORG_ID_15", ""))
    label_safe = scrub(label)
    reason_safe = scrub(reason)

    block = (
        "=== RESULT ===\n"
        "STATUS=INVALID_INPUT\n"
        f"ERROR_DETAIL={label_safe}: {reason_safe}\n"
        f"AGENT_API_NAME={agent_api_name}\n"
        f"ORG_ID_18={org_id_18}\n"
        f"ORG_ID_15={org_id_15}\n"
    )

    # Default ERROR_TEE — skill-scoped data dir. Runtime-agnostic default
    # mirrors runtime.resolve_data_root() in the sibling runtime module
    # (duplicated rather than imported because fs_guard.py is also invoked
    # standalone from SKILL.md bash, where the override hook isn't active).
    tee_default = str(
        pathlib.Path.home()
        / ".vibe"
        / "data"
        / "investigating-agentforce-architecture"
        / "_last_error_result.txt"
    )
    tee_path = os.environ.get("ERROR_TEE") or tee_default
    try:
        p = pathlib.Path(tee_path)
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(block)
    except OSError:
        pass

    sys.stdout.write(block)
    sys.exit(1)


def check_symlink(value: str, label: str) -> None:
    if pathlib.Path(value).is_symlink():
        emit_failure(f"path is a symlink (refusing to follow): {value}", label)


def check_owned(value: str, label: str) -> None:
    p = pathlib.Path(value)
    if not p.exists():
        return
    try:
        st = p.stat()
    except OSError as e:
        emit_failure(f"stat failed: {e}", label)
        return
    if st.st_uid != os.getuid():
        emit_failure(f"foreign-owned (uid {st.st_uid} != current {os.getuid()}): {value}", label)


def check_uuid(value: str, label: str) -> None:
    if not UUID_RE.match(value):
        emit_failure("does not match UUID pattern (8-4-4-4-12 lowercase hex)", label)


def check_org_id_15(value: str, label: str) -> None:
    if not ORG_ID_15_RE.match(value):
        emit_failure("must be exactly 15 alphanumeric characters", label)


def check_api_name(value: str, label: str) -> None:
    if not API_NAME_RE.match(value):
        emit_failure("does not match [A-Za-z0-9_]+ (Salesforce API name rules)", label)


def check_api_version(value: str, label: str) -> None:
    # api_version must be `vNN.N` — no slashes, no dots beyond the one,
    # no path-traversal sequences.
    if not API_VERSION_RE.match(value):
        emit_failure("does not match v<major>.<minor> (e.g. v60.0)", label)


def check_agent_version(value: str, label: str) -> None:
    # Agent version must be `v<digits>` — no dots, no slashes.
    if not AGENT_VERSION_RE.match(value):
        emit_failure("does not match v<digits> (e.g. v5)", label)


def check_not_empty(value: str, label: str) -> None:
    if not value:
        emit_failure("must not be empty", label)


CHECKS = {
    "symlink": check_symlink,
    "owned": check_owned,
    "uuid": check_uuid,
    "org_id_15": check_org_id_15,
    "api_name": check_api_name,
    "api_version": check_api_version,
    "agent_version": check_agent_version,
    "not_empty": check_not_empty,
}


def main() -> int:
    if len(sys.argv) != 4:
        sys.stdout.write(
            "=== RESULT ===\n"
            "STATUS=INVALID_INPUT\n"
            "ERROR_DETAIL=fs_guard internal: wrong argv count (need value, label, check)\n"
        )
        return 1
    value, label, check = sys.argv[1], sys.argv[2], sys.argv[3]

    for arg in (value, label, check):
        if any(c == "\0" or (ord(c) < 0x20 and c not in "\t\n") for c in arg):
            emit_failure("argv contains control characters", label or "argv")

    if check not in VALID_CHECKS:
        emit_failure(f"unknown check type '{check}' (valid: {', '.join(sorted(VALID_CHECKS))})", label)

    CHECKS[check](value, label)
    return 0


if __name__ == "__main__":
    sys.exit(main())

"""Canonical path helpers for investigating-agentforce-architecture.

Layout:

    ~/.vibe/data/investigating-agentforce-architecture/
    └── <org_id_15>/
        └── <agent_api_name>__<agent_version>/
            ├── <agent>_<ver>_metadata_tree.json     ← rendered tree
            └── .emit_ctx.json                       ← per-run shell ctx

Validation strategy
-------------------
The three "agent-identity" segments (org_id_15, agent_api_name, agent_version)
are validated by the regex helpers in the sibling ``fs_guard`` module. Any
segment containing ``..``, ``/``, or characters outside its regex charclass
raises ``PathValidationError`` — direct ``Path`` composition from unvalidated
input would be a path-traversal vulnerability and is prohibited.

Runtime override
----------------
``DATA_ROOT`` / ``CACHE_ROOT`` are resolved via the ``runtime`` module so
that entry scripts can override them via ``--data-dir`` / ``--cache-dir``
CLI flags. The default values (``~/.vibe/{data,cache}/...``)
are runtime-agnostic; AFV OOTB, Codex, Cursor, OpenCode each pass their
own override paths.
"""
from __future__ import annotations

from pathlib import Path

from . import fs_guard
from . import runtime


# -----------------------------------------------------------------------------
# Roots
# -----------------------------------------------------------------------------

DATA_ROOT: Path = runtime.resolve_data_root("investigating-agentforce-architecture")
CACHE_ROOT: Path = runtime.resolve_cache_root("investigating-agentforce-architecture")


class PathValidationError(ValueError):
    """Raised when a path segment fails validation.

    Wraps fs_guard's ``ValidationError`` so callers have a single exception
    type to catch.
    """

    def __init__(self, label: str, reason: str) -> None:
        self.label = label
        self.reason = reason
        super().__init__(f"{label}: {reason}")


# -----------------------------------------------------------------------------
# Validation helpers
# -----------------------------------------------------------------------------


def _validate_agent_triple(
    org_id_15: str, agent_api_name: str, agent_version: str
) -> None:
    """Validate the three identity segments. Raises PathValidationError on
    failure, wrapping fs_guard's ValidationError so callers don't need to
    know about fs_guard internals."""
    try:
        fs_guard.validate_org_id_15(org_id_15, label="org_id_15")
        fs_guard.validate_api_name(agent_api_name, label="agent_api_name")
        fs_guard.validate_agent_version(agent_version, label="agent_version")
    except fs_guard.ValidationError as e:
        raise PathValidationError(e.label, e.reason) from e


# -----------------------------------------------------------------------------
# Path builders
# -----------------------------------------------------------------------------


def agent_dir(
    org_id_15: str, agent_api_name: str, agent_version: str
) -> Path:
    """Return ``DATA_ROOT/<org_id_15>/<agent_api_name>__<agent_version>/``.

    All three segments are regex-validated before being joined.
    """
    _validate_agent_triple(org_id_15, agent_api_name, agent_version)
    return DATA_ROOT / org_id_15 / f"{agent_api_name}__{agent_version}"


def architecture_tree_path(
    org_id_15: str, agent_api_name: str, agent_version: str
) -> Path:
    """Return the path to the rendered metadata tree JSON.

    Filename convention: ``<agent_api_name>_<agent_version>_metadata_tree.json``.
    """
    base = agent_dir(org_id_15, agent_api_name, agent_version)
    return base / f"{agent_api_name}_{agent_version}_metadata_tree.json"


def architecture_emit_ctx_path(
    org_id_15: str, agent_api_name: str, agent_version: str
) -> Path:
    """Return the path to the per-run ``.emit_ctx.json``.

    The file is dotfile-named to signal per-run shell state.
    """
    base = agent_dir(org_id_15, agent_api_name, agent_version)
    return base / ".emit_ctx.json"

"""Runtime override hook for DATA_ROOT / CACHE_ROOT.

Default layout is ``~/.vibe/data/<skill>`` and
``~/.vibe/cache/<skill>`` — runtime-agnostic branding that
follows the repo's "tool-agnostic language" guideline.

Other runtimes (AFV OOTB, Codex, Cursor, OpenCode) can override via
``--data-dir`` / ``--cache-dir`` CLI flags on entry scripts. Those flags
call ``set_*_override()`` BEFORE any pipeline module imports ``paths.py``,
so the resolution helpers below pick up the override.

Three-level rebind contract
---------------------------
Python's ``from X import Y`` snapshots ``Y`` into the importer's local
namespace. Mutating ``X.Y`` later does NOT update the importer's local
binding. Entry scripts that capture ``DATA_ROOT`` / ``CACHE_ROOT`` at
module top must therefore rebind in THREE places after override:

1. ``paths.DATA_ROOT`` — the source of truth
2. ``config.DATA_ROOT`` — the re-export
3. The entry script's own local ``DATA_ROOT`` (via ``global``)
"""
from __future__ import annotations

from pathlib import Path
from typing import Optional

_DATA_OVERRIDE: Optional[Path] = None
_CACHE_OVERRIDE: Optional[Path] = None


def set_data_root_override(p: Optional[Path]) -> None:
    """Set the data-root override. Pass ``None`` to clear."""
    global _DATA_OVERRIDE
    _DATA_OVERRIDE = p


def set_cache_root_override(p: Optional[Path]) -> None:
    """Set the cache-root override. Pass ``None`` to clear."""
    global _CACHE_OVERRIDE
    _CACHE_OVERRIDE = p


def resolve_data_root(skill_name: str) -> Path:
    """Resolve the data root for ``skill_name``.

    Returns the override if set, else the runtime-agnostic default
    ``~/.vibe/data/<skill_name>``.
    """
    return _DATA_OVERRIDE or (Path.home() / ".vibe" / "data" / skill_name)


def resolve_cache_root(skill_name: str) -> Path:
    """Resolve the cache root for ``skill_name``.

    Returns the override if set, else the runtime-agnostic default
    ``~/.vibe/cache/<skill_name>``.
    """
    return _CACHE_OVERRIDE or (Path.home() / ".vibe" / "cache" / skill_name)

"""SQL-escaping helpers for SOQL string literals.
"""
from __future__ import annotations


def _escape_sql_literal(s: str) -> str:
    """Double single quotes per SOQL escaping rule. Handles O'Brien →
    O''Brien, `'; DROP --` → `''; DROP --` (still harmless because it's
    wrapped in surrounding single quotes)."""
    return s.replace("'", "''")

#!/usr/bin/env python3
"""Cache hit/miss check for investigating-agentforce-architecture.

Replaces old agent Phase 0.5 (manifest read, schema-version check, TTL age
check, path-existence check, env export on hit, RESULT-block assembly).

Flow:
  1. If FORCE_REFRESH=true → emit CACHE_HIT=false, exit 0.
  2. Read $CACHE_DIR/manifest.json. Any I/O or parse failure → miss.
  3. if schema_version != config.SCHEMA_VERSION → miss,
     emit CACHE_INVALIDATED_REASON=schema-version-mismatch, and
     `shutil.rmtree` the cache dir (gated to paths under CACHE_ROOT).
  4. If data_path file missing → miss (atomic rename failed mid-write).
  5. If age > ttl_days (default 7) → miss (no rmtree — just stale).
  6. Hit path:
     - Copy {agent}_{ver}_metadata_tree.json from cache into DATA_DIR +
       WORK_DIR.
     - Copy `declared_action_tree.json` sidecar into WORK_DIR so emit ctx
       + downstream greps work.
     - Stdout: eval-able lines (CACHE_HIT=true, CACHED_AT_UTC, NODE_COUNT,
       DEPTH, AGENT_GENERATION, BOT_ID, VERSION_AUTO_PICKED, PARTIAL,
       UNRESOLVED_COUNT, OUTPUT_JSON_PATH, OUTPUT_SUMMARY_PATH).

`{tree_base}.summary.md` was dropped from the output contract.
Cache-hit no longer copies or regenerates a summary file; OUTPUT_SUMMARY_PATH
is emitted empty for RESULT-block shape stability.

Usage:
    eval "$(python3 cache_check.py)"

Inputs (env):
    CACHE_DIR        required — $CACHE_ROOT/$ORG_ID_15/$AGENT__$VER
    DATA_DIR         required — $DATA_ROOT/$ORG_ID_15/$AGENT__$VER
    WORK_DIR         required
    AGENT_API_NAME   required
    AGENT_VERSION    required
    FORCE_REFRESH    'true'|'false' (default false)

Outputs:
    stdout: eval-able K=V lines
    files (hit only): DATA_DIR/{tree_base}.json and matching copy in WORK_DIR
    exit 0 always (miss is not an error)
"""
import json
import os
import pathlib
import shlex
import shutil
import sys
import datetime as dt

import config  # for SCHEMA_VERSION + CACHE_ROOT path-guard 


# refuse to operate if CACHE_ROOT itself is a symlink.
# _safe_rmtree_under_cache_root calls resolve() on both target and root; a
# symlinked CACHE_ROOT collapses the is_relative_to check because both sides
# land on the symlink target, so rmtree would happily delete whatever lives
# there — outside the sanctioned cache subtree. The guard was designed to
# prevent rmtree escaping the cache; if the root itself can escape, the
# guard is moot. Fail fast at import time so the failure surfaces at the
# correct layer (configuration, not deletion).
#
# Path.is_symlink() returns False for non-existent paths, which is the
# correct behaviour for pristine installs (CACHE_ROOT hasn't been created
# yet; the script will mkdir it on first write and subsequent imports will
# validate the real directory).
if config.CACHE_ROOT.is_symlink():
    raise RuntimeError(
        f"CACHE_ROOT is a symlink ({config.CACHE_ROOT} -> "
        f"{config.CACHE_ROOT.resolve()}). This is rejected for safety; the "
        "_safe_rmtree_under_cache_root guard cannot protect against a "
        "symlinked-root escape. Resolve the symlink or update "
        "config.CACHE_ROOT to a real directory."
    )


def miss(reason: str | None = None):
    """Emit CACHE_HIT=false and exit 0.

    if `reason` is provided, emit a second K=V line
    `CACHE_INVALIDATED_REASON=<reason>` so downstream emit_result can
    surface the invalidation cause in the RESULT block.
    """
    sys.stdout.write("CACHE_HIT=false\n")
    if reason:
        sys.stdout.write(f"CACHE_INVALIDATED_REASON={shlex.quote(reason)}\n")
    sys.exit(0)


def _safe_rmtree_under_cache_root(target: pathlib.Path) -> bool:
    """`shutil.rmtree(target)` ONLY if `target` resolves under CACHE_ROOT.

    Defence against a miscomputed cache_dir that could point outside the
    sanctioned cache root (symlink shenanigans, env-var tampering, or a
    future refactor that forgets to run through `build_agent_cache_dir`).
    We resolve both sides to absolute paths and confirm CACHE_ROOT is a
    prefix before handing to rmtree.

    Returns True on successful deletion, False if the path was refused or
    the rmtree itself failed. Never raises — callers on the miss() path
    can't act on a failure anyway.
    """
    try:
        target_abs = target.resolve(strict=False)
        root_abs = config.CACHE_ROOT.resolve(strict=False)
    except (OSError, RuntimeError):
        return False
    # Path.is_relative_to exists on 3.9+. Explicit check is clearer than
    # a try/except on relative_to().
    try:
        if not target_abs.is_relative_to(root_abs):
            return False
    except AttributeError:  # pragma: no cover — safety for <3.9
        try:
            target_abs.relative_to(root_abs)
        except ValueError:
            return False
    if not target_abs.exists():
        return True  # nothing to do, treat as successful no-op
    try:
        shutil.rmtree(target_abs)
    except OSError:
        return False
    return True


def main() -> int:
    try:
        cache_dir = pathlib.Path(os.environ["CACHE_DIR"])
        data_dir = pathlib.Path(os.environ["DATA_DIR"])
        work_dir = pathlib.Path(os.environ["WORK_DIR"])
        agent_api_name = os.environ["AGENT_API_NAME"]
        agent_version = os.environ["AGENT_VERSION"]
    except KeyError as e:
        sys.stderr.write(f"cache_check.py: missing env {e}\n")
        miss()

    if (os.environ.get("FORCE_REFRESH", "").strip().lower() == "true"):
        miss()

    manifest_path = cache_dir / "manifest.json"
    if not manifest_path.is_file():
        miss()

    try:
        manifest = json.loads(manifest_path.read_text())
    except (OSError, json.JSONDecodeError):
        miss()

    # Schema version gate.
    # strict match against config.SCHEMA_VERSION. Any mismatch (too
    # old OR unexpected future/unknown value) invalidates the cache, and
    # we delete the cache directory so the next run starts clean — a
    # stale tree under a legacy schema is worse than no cache, because
    # downstream code may parse shapes it no longer understands. The
    # legacy `< 2.4` check is subsumed (anything that doesn't equal the
    # current SCHEMA_VERSION triggers this branch).
    schema = str(manifest.get("schema_version") or "0")
    if schema != config.SCHEMA_VERSION:
        _safe_rmtree_under_cache_root(cache_dir)
        miss("schema-version-mismatch")

    # data_path existence
    data_path_s = manifest.get("data_path") or ""
    data_path = pathlib.Path(data_path_s) if data_path_s else None
    if not data_path or not data_path.is_file():
        miss()

    # TTL age check
    try:
        built = dt.datetime.fromisoformat(
            (manifest.get("built_at_utc") or "").replace("Z", "+00:00")
        )
    except ValueError:
        miss()
    age_days = (dt.datetime.now(dt.timezone.utc) - built).days
    ttl = int(manifest.get("ttl_days") or 7)
    if age_days > ttl:
        miss()

    # --- Hit path ---
    tree_base = f"{agent_api_name}_{agent_version}_metadata_tree"
    data_dir.mkdir(parents=True, exist_ok=True)
    work_dir.mkdir(parents=True, exist_ok=True)

    dst_json = data_dir / f"{tree_base}.json"

    # The manifest's data_path points at the authoritative tree copy. Re-copy
    # to DATA_DIR + WORK_DIR so the filesystem layout is stable for callers.
    try:
        shutil.copy(data_path, dst_json)
    except OSError:
        miss()

    # Stage a declared_action_tree.json sidecar in WORK_DIR for any downstream
    # script that expects the generic name.
    try:
        shutil.copy(dst_json, work_dir / "declared_action_tree.json")
    except OSError:
        pass

    # no summary.md handling — dropped from the output contract.
    agent_meta = manifest.get("agent") or {}
    kind_counts = manifest.get("kind_counts") or {}

    exports = [
        ("CACHE_HIT", "true"),
        ("CACHED_AT_UTC", manifest.get("built_at_utc", "")),
        ("NODE_COUNT", str(manifest.get("node_count", 0))),
        ("DEPTH", str(manifest.get("depth", 0))),
        ("AGENT_GENERATION", agent_meta.get("generation") or "unknown"),
        ("BOT_ID", agent_meta.get("bot_id") or ""),
        ("BOT_MASTER_LABEL", agent_meta.get("master_label") or ""),
        ("VERSION_AUTO_PICKED", "true" if agent_meta.get("_version_auto_picked") else "false"),
        ("PARTIAL", "true" if manifest.get("partial") else "false"),
        ("UNRESOLVED_COUNT", str(manifest.get("unresolved_count", 0))),
        ("OUTPUT_JSON_PATH", str(dst_json)),
        # summary.md dropped from the output contract; field kept
        # empty for RESULT-block shape stability.
        ("OUTPUT_SUMMARY_PATH", ""),
        ("PLANNER_NAME", agent_meta.get("planner_name") or ""),
    ]
    # Pass kind counts too, in case finalize.py runs later without re-parsing
    # (it won't on cache hit, but the values keep the RESULT block complete).
    for k, v in kind_counts.items():
        exports.append((f"KC_{k}", str(v)))

    sys.stdout.write("\n".join(f"{k}={shlex.quote(v)}" for k, v in exports) + "\n")
    return 0


if __name__ == "__main__":
    sys.exit(main())

"""Shared paths + constants for investigating-agentforce-architecture.

Resolves SKILL_ROOT relative to this file's location — works on every
runtime without env vars. SKILL.md bash blocks still read the
``PLUGIN_ROOT`` env var (with a default fallback) because bash has no
``__file__`` equivalent.

Path layout:

    ~/.vibe/data/investigating-agentforce-architecture/
    └── <org_id_15>/
        └── <agent_api_name>__<agent_version>/        ← architecture artifacts

Every path component embedded in a cache / data directory must be
regex-validated before being joined to a Path. Four components are in
scope: `api_version`, `org_id15`, `agent_api_name`, `agent_version`.
(`planner_name` is validated separately at the SOQL substitution
boundary by `soql_loader.load_soql` — it never appears in a filesystem
path.) The helpers below are the ONLY sanctioned way to build those
paths — direct Path composition from unvalidated strings is a
path-traversal vulnerability.
"""
from __future__ import annotations

import sys
from pathlib import Path

# SKILL_ROOT is the directory holding SKILL.md, derived from this file's location:
#   <SKILL_ROOT>/scripts/config.py  →  Path(__file__).resolve().parent.parent
# Python doesn't read PLUGIN_ROOT — it's only for SKILL.md bash blocks
# (PR3 will replace those with a proper entry script).
SKILL_ROOT = Path(__file__).resolve().parent.parent
SOQL_DIR = SKILL_ROOT / "assets" / "soql"
CLI_DIR = SKILL_ROOT / "assets" / "cli"
MERMAID_DIR = SKILL_ROOT / "assets" / "mermaid"
REFERENCES_DIR = SKILL_ROOT / "references"


# -----------------------------------------------------------------------------
# Shared path helpers — sourced from scripts/_shared/.
# -----------------------------------------------------------------------------

_SCRIPTS_DIR = Path(__file__).resolve().parent
if str(_SCRIPTS_DIR) not in sys.path:
    sys.path.insert(0, str(_SCRIPTS_DIR))

from _shared import paths as _paths  # type: ignore  # noqa: E402
from _shared import fs_guard  # type: ignore  # noqa: E402,F401 — re-exported for sibling scripts + tests

DATA_ROOT = _paths.DATA_ROOT
CACHE_ROOT = _paths.CACHE_ROOT
PROBE_CACHE_ROOT = CACHE_ROOT / "_channel_probe"

# Cache TTL — 7 days. Probe TTL same.
CACHE_TTL_DAYS = 7
PROBE_TTL_DAYS = 7

# BFS defensive termination guard.
#
# Historically this was `5` and acted as a functional constraint on
# chain depth. That was wrong: shared utility flows like `handleFlowFault`
# appear on every real flow's fault path, so any moderately nested tree
# (`[FLOW] A → [FLOW] B → [FLOW] handleFlowFault`) tripped the cap and
# surfaced the utility in `_pending_fetches` even though it is trivially
# expandable. Cycle detection now runs via a per-branch ancestor path set
# (`visited_in_path` in `inflate_flow_leaf`), which is the textbook-correct
# primitive for this problem: the same flow visited on two sibling branches
# is not a cycle, but the same flow visited along its own ancestor chain is.
#
# This constant is retained purely as a defensive last-resort termination
# guard for pathological graphs that somehow evade per-branch cycle detection
# (e.g. a bug regression in `_cycle_key`). In practice the per-branch set
# terminates every real bot tree long before this depth is reached.
MAX_BFS_DEPTH = 20

# Schema version — 3.1 (2026-05-05) canonicalizes `invocation_type` on
# STANDARD_ACTION nodes, formerly split across `raw_invocation_type` and
# `raw_action_type`. Must match `parse_wave.init_tree`'s `_schema_version`
# literal. Any bump forces a cache rebuild via `cache_check.py`'s gate.
SCHEMA_VERSION = "3.1"

# Default parallelism for ThreadPoolExecutor (Flow metadata fan-out).
DEFAULT_PARALLELISM = 5


# -----------------------------------------------------------------------------
# validated path builders
# -----------------------------------------------------------------------------
# fs_guard is re-exported above from `scripts/_shared/`.
# Sibling scripts (soql_loader, render_architecture, fetch_soql)
# and the test suite import it as ``from config import fs_guard`` so the
# resolution dance lives in one place.


def build_agent_data_dir(org_id_15: str, agent_api_name: str, agent_version: str) -> Path:
    """Return DATA_ROOT/<org_id_15>/<agent_api_name>__<agent_version>/.

    Thin wrapper over ``_shared.paths.agent_dir`` — kept so existing
    callers don't need to change import sites. The shared helper raises
    ``paths.PathValidationError`` (a ValueError subclass); we re-raise
    the underlying ``fs_guard.ValidationError`` so existing tests that
    catch that specific type keep passing.
    """
    fs_guard.validate_org_id_15(org_id_15, label="org_id_15")
    fs_guard.validate_api_name(agent_api_name, label="agent_api_name")
    fs_guard.validate_api_name(agent_version, label="agent_version")
    return _paths.DATA_ROOT / org_id_15 / f"{agent_api_name}__{agent_version}"


def build_agent_cache_dir(org_id_15: str, agent_api_name: str, agent_version: str) -> Path:
    """Return CACHE_ROOT/<org_id_15>/<agent_api_name>__<agent_version>/.

    Mirrors ``build_agent_data_dir`` under the cache root. Validation gate
    matches — every segment regex-checked before the join.
    """
    fs_guard.validate_org_id_15(org_id_15, label="org_id_15")
    fs_guard.validate_api_name(agent_api_name, label="agent_api_name")
    fs_guard.validate_api_name(agent_version, label="agent_version")
    return CACHE_ROOT / org_id_15 / f"{agent_api_name}__{agent_version}"


def build_probe_cache_dir(org_id_15: str, api_version: str) -> Path:
    """Return PROBE_CACHE_ROOT/<org_id_15>/<api_version>/.

    org_id_15 via validate_api_name (regex tolerates 15-char alnum),
    api_version via validate_api_version (enforces `vNN.N` shape — rejects
    `..`, `/`, and any non-version string).
    """
    fs_guard.validate_org_id_15(org_id_15, label="org_id_15")
    fs_guard.validate_api_version(api_version, label="api_version")
    return PROBE_CACHE_ROOT / org_id_15 / api_version

#!/usr/bin/env python3
"""Finalize: write DATA_DIR + CACHE_DIR atomically, build manifest.

Replaces old agent Phase 7. Steps:
  1. Load $WORK_DIR/declared_action_tree.json.
  2. Compute _partial = !(planner_ok && _pending_fetches empty); strip _visited.
  3. Stage DATA_DIR at $DATA_DIR.tmp: copy tree as {api}_{ver}_metadata_tree.json.
  4. rmtree final DATA_DIR; rename .tmp → DATA_DIR.
  5. Stage CACHE_DIR at $CACHE_DIR.tmp: copy metadata/<wave> dirs + sidecars.
  6. Write manifest.json.
  7. rmtree final CACHE_DIR; rename .tmp → CACHE_DIR.
  8. Seed .gitignore on data_root + cache_root parents.
  9. Write $WORK_DIR/.built_at.txt (ISO-Z UTC).

the previously-emitted `{api}_{ver}_metadata_tree.summary.md` is
dropped from the output contract — it was a redundant summary of the tree
JSON. Consumers should read the JSON directly.

Usage:
    python3 finalize.py

Inputs (env):
    WORK_DIR       required
    CACHE_DIR      required
    DATA_DIR       required
    PLANNER_NAME   optional — empty → tree marked partial
    AGENT_API_NAME, AGENT_VERSION — used for filenames

Outputs:
    $DATA_DIR/{api}_{ver}_metadata_tree.json
    $DATA_DIR/last_built_at.txt
    $CACHE_DIR/manifest.json + metadata/<wave>/... + parsed sidecars
    $WORK_DIR/.built_at.txt
    exit 0 success, 1 on any write failure → caller emits STATUS=WRITE_FAILED
"""
import datetime
import json
import os
import pathlib
import shutil
import sys


def sort_tree_in_place(root: dict) -> None:
    """Sort `root.children` and each TOPIC's children alphabetically.

    Ordering rules (2026-05-05, schema 3.1):
    - `BOT_DEFINITION.children`: TOPIC nodes first (by `api_name`
      case-insensitive), then non-topic plannerActions (by kind, then
      `api_name`). This preserves the "planner-level actions as a
      distinct trailing group" convention for readers who scan the
      rendered tree top-down.
    - Each TOPIC's children: alphabetical by `api_name`
      case-insensitive.
    - FLOW children are NOT sorted — flow-actionCall / subflow order
      is the flow author's execution sequence, not a set.
    - GEN_AI_FUNCTION / APEX / PROMPT_TEMPLATE / STANDARD_ACTION leaves
      don't have children that warrant sorting.

    Applied as a final pass after tree assembly and before the
    authoritative JSON write, so ordering is pinned at the single
    source of truth — the renderer doesn't re-sort.
    """
    if not isinstance(root, dict):
        return
    children = root.get("children") or []
    if not children:
        return

    def _root_sort_key(node: dict) -> tuple:
        kind = node.get("kind") or ""
        api = (node.get("api_name") or "").casefold()
        # TOPIC first (tier 0), everything else second (tier 1). Within
        # each tier, stable by kind + api_name case-insensitive.
        tier = 0 if kind == "TOPIC" else 1
        return (tier, kind, api)

    root["children"] = sorted(children, key=_root_sort_key)

    for child in root["children"]:
        if child.get("kind") == "TOPIC":
            child_children = child.get("children") or []
            child["children"] = sorted(
                child_children,
                key=lambda n: (n.get("api_name") or "").casefold(),
            )


def main() -> int:
    try:
        work_dir = pathlib.Path(os.environ["WORK_DIR"])
        cache_dir = pathlib.Path(os.environ["CACHE_DIR"])
        data_dir = pathlib.Path(os.environ["DATA_DIR"])
        agent_api_name = os.environ["AGENT_API_NAME"]
        agent_version = os.environ["AGENT_VERSION"]
    except KeyError as e:
        sys.stderr.write(f"finalize.py: missing env {e}\n")
        return 1

    planner_name = os.environ.get("PLANNER_NAME", "")
    tree_path = work_dir / "declared_action_tree.json"

    try:
        tree = json.loads(tree_path.read_text())
    except (OSError, json.JSONDecodeError) as e:
        sys.stderr.write(f"finalize.py: cannot read {tree_path}: {e}\n")
        return 1

    # Compute _partial. _pending_fetches drains as wave-B BFS resolves
    # refs; failures (HTTP 4xx, iteration-cap exhaustion, managed-flow
    # filter mismatch) move OUT of _pending_fetches and into _unresolved.
    # If we only check _pending_fetches we'd silently call a run with
    # _unresolved entries "converged" — STATUS=OK with hidden failures.
    # Both buckets must be empty for a clean run.
    planner_ok = bool(planner_name)
    pending_total = sum(len(v) for v in (tree.get("_pending_fetches") or {}).values())
    unresolved_count = len(tree.get("_unresolved") or [])
    waves_converged = pending_total == 0 and unresolved_count == 0
    tree["_partial"] = not (planner_ok and waves_converged)

    # Bug F fix: parse_wave + main set _partial_reason from
    # _pending_fetches only (legacy predicate). When _pending is empty
    # but _unresolved is non-empty, those writers set reason=None. The
    # promotion above flips _partial=True from the unresolved bucket, so
    # the reason needs a matching promotion or PARTIAL_REASON= ends up
    # blank in the RESULT block.
    if tree["_partial"] and not tree.get("_partial_reason"):
        if not planner_ok:
            tree["_partial_reason"] = "no-planner"
        elif unresolved_count > 0:
            tree["_partial_reason"] = "unresolved-refs"
        else:
            tree["_partial_reason"] = "pending-refs"

    # Strip _visited (internal state — not part of the durable artifact)
    tree.pop("_visited", None)

    # Pin deterministic child ordering as the last assembly step before
    # the authoritative write — see `sort_tree_in_place` docstring for
    # the ordering rules. Downstream readers (render_architecture,
    # summarize_tree, third-party tooling) therefore see a single
    # canonical order from disk; they don't re-sort.
    sort_tree_in_place(tree.get("root") or {})

    # Rewrite the authoritative tree before copying it anywhere else
    try:
        tmp = tree_path.with_suffix(tree_path.suffix + ".tmp")
        tmp.write_text(json.dumps(tree, indent=2))
        os.replace(tmp, tree_path)
    except OSError as e:
        sys.stderr.write(f"finalize.py: cannot rewrite tree: {e}\n")
        return 1

    built_at = datetime.datetime.now(datetime.timezone.utc).isoformat().replace("+00:00", "Z")
    a = tree.get("agent", {}) or {}
    tree_base = f"{agent_api_name}_{agent_version}_metadata_tree"

    # --- DATA_DIR staging ---
    # summary.md dropped from the output contract; DATA_DIR now
    # holds only the tree JSON + last_built_at.txt (+ whatever the caller
    # writes alongside, e.g. architecture.md from render_architecture).
    #
    # The prior pattern was `rmtree(data_dir); data_tmp.rename(data_dir)` —
    # destructive of any co-tenant content under `<agent>__<ver>/`. We
    # now iterate staging's children and overwrite each one into data_dir
    # individually, leaving unrelated siblings intact. See
    # `main.py:_swap_dir_atomic` for the production path with the same
    # invariant.
    data_tmp = data_dir.with_suffix(".tmp")
    try:
        if data_tmp.exists():
            shutil.rmtree(data_tmp)
        data_tmp.mkdir(parents=True)
        shutil.copy(tree_path, data_tmp / f"{tree_base}.json")

        (data_tmp / "last_built_at.txt").write_text(built_at + "\n")

        data_dir.mkdir(parents=True, exist_ok=True)
        for staged in list(data_tmp.iterdir()):
            target = data_dir / staged.name
            if target.exists() or target.is_symlink():
                if target.is_dir() and not target.is_symlink():
                    shutil.rmtree(target)
                else:
                    target.unlink()
            shutil.move(str(staged), str(target))
        shutil.rmtree(data_tmp, ignore_errors=True)
    except OSError as e:
        sys.stderr.write(f"finalize.py: DATA_DIR write failed: {e}\n")
        try:
            if data_tmp.exists():
                shutil.rmtree(data_tmp)
        except OSError:
            pass
        return 1

    # --- CACHE_DIR staging ---
    cache_tmp = cache_dir.with_suffix(".tmp")
    try:
        if cache_tmp.exists():
            shutil.rmtree(cache_tmp)
        cache_tmp.mkdir(parents=True)

        # Mirror sf_meta/<wave>/ into metadata/<wave>/ verbatim so cache has:
        # retrieve.json (sf CLI output — critical for debugging silent
        # retrieves that produced zero members)
        # unpackaged/ (parsed XML tree)
        # unpackaged.zip is intentionally skipped — it's byte-for-byte the
        # same as unpackaged/ and doubles cache size on every run.
        meta_dir = cache_tmp / "metadata"
        meta_dir.mkdir()
        sf_meta = work_dir / "sf_meta"
        if sf_meta.exists():
            for wave_dir in sorted(sf_meta.iterdir()):
                if not wave_dir.is_dir():
                    continue
                dst = meta_dir / wave_dir.name
                dst.mkdir(parents=True, exist_ok=True)
                for item in wave_dir.iterdir():
                    if item.name == "unpackaged.zip":
                        continue  # redundant with unpackaged/
                    if item.is_dir():
                        shutil.copytree(item, dst / item.name, dirs_exist_ok=True)
                    else:
                        shutil.copy(item, dst / item.name)

        # Copy every WORK_DIR top-level artifact except:
        # - .emit_ctx.json (per-run shell state, irrelevant to cache)
        # - .built_at.txt (recorded in manifest.built_at_utc)
        # - sf_meta/ (handled above)
        # This catches declared_action_tree.json, _bundle_parsed.json,
        # _agent_generation.txt, _bot_definition.json, _bot_versions.json,
        # and anything future scripts write as top-level sidecars.
        SKIP_NAMES = {".emit_ctx.json", ".built_at.txt", "sf_meta"}
        for item in work_dir.iterdir():
            if item.name in SKIP_NAMES:
                continue
            if item.name.endswith(".tmp"):
                continue  # atomic-write staging from some other script
            if item.is_dir():
                shutil.copytree(item, cache_tmp / item.name, dirs_exist_ok=True)
            elif item.is_file():
                shutil.copy(item, cache_tmp / item.name)

        manifest = {
            "built_at_utc": built_at,
            "schema_version": tree.get("_schema_version", "2.4"),
            "agent": a,
            "node_count": tree.get("node_count", 0),
            "depth": tree.get("depth", 0),
            "kind_counts": tree.get("_kind_counts", {}),
            "ttl_days": 7,
            "data_path": str(data_dir / f"{tree_base}.json"),
            "partial": tree.get("_partial", False),
            "unresolved_count": len(tree.get("_unresolved", []) or []),
        }
        (cache_tmp / "manifest.json").write_text(json.dumps(manifest, indent=2))

        if cache_dir.exists():
            shutil.rmtree(cache_dir)
        cache_tmp.rename(cache_dir)
    except OSError as e:
        sys.stderr.write(f"finalize.py: CACHE_DIR write failed: {e}\n")
        try:
            if cache_tmp.exists():
                shutil.rmtree(cache_tmp)
        except OSError:
            pass
        return 1

    # --- .gitignore seeding on parent dirs (data_root + cache_root) ---
    for root in (cache_dir.parent.parent, data_dir.parent.parent):
        try:
            gi = root / ".gitignore"
            if not gi.exists():
                root.mkdir(parents=True, exist_ok=True)
                gi.write_text("*\n")
        except OSError:
            pass

    # --- .built_at.txt for emit ctx ---
    try:
        (work_dir / ".built_at.txt").write_text(built_at + "\n")
    except OSError as e:
        sys.stderr.write(f"finalize.py: .built_at.txt write failed: {e}\n")
        # Non-fatal — finalize succeeded; emit can recompute

    print(f"[finalize] built_at: {built_at}")
    print(f"[finalize] data  written: {data_dir}/{tree_base}.json")
    print(f"[finalize] cache written: {cache_dir}")
    return 0


if __name__ == "__main__":
    sys.exit(main())