github · antigravity-workspace-template6f1c98821e32

Evidence

A short confirmation message.

    """

    print(f"DEBUG: Sending email to {to}. Body length: {len(body)}")

    return f"Email sent to {to} (mock)."

Remediation

Strip C0/C1 control codes before printing user-controlled values. Python: re.sub(r"[\x00-\x08\x0b-\x1f\x7f]", "", s). Prefer a structured logger (json/logfmt) over raw print to stdout.

Evidence

# For now, I will return a placeholder string.

    # </thought>

    print(f"DEBUG: Performing web search for '{query}'")

    # Mock response

    results = f"Search results for: {query}\n1. Result A for {query}...\n2. Result B for {query}..."

Remediation

Strip C0/C1 control codes before printing user-controlled values. Python: re.sub(r"[\x00-\x08\x0b-\x1f\x7f]", "", s). Prefer a structured logger (json/logfmt) over raw print to stdout.

Evidence

scan_report_path = ag_dir / "scan_report.json"

    scan_payload = _build_scan_payload(report)

    print("[1/3] Scan payload built; writing scan_report.json...", file=sys.stderr)

    scan_report_path.write_text(

        json.dumps(scan_payload, ensure_ascii=False, indent=2),

        encoding="utf-8",

Remediation

Strip C0/C1 control codes before printing user-controlled values. Python: re.sub(r"[\x00-\x08\x0b-\x1f\x7f]", "", s). Prefer a structured logger (json/logfmt) over raw print to stdout.

Evidence

# Track agent changes

        elif event.type == "agent_updated_stream_event":

            print(

                f"[stream] → Agent: {event.new_agent.name}",

                file=sys.stderr,

                flush=True,

            )

        # Handle generated items

        elif event.type == "run_item_stream_event":

Remediation

Strip C0/C1 control codes before printing user-controlled values. Python: re.sub(r"[\x00-\x08\x0b-\x1f\x7f]", "", s). Prefer a structured logger (json/logfmt) over raw print to stdout.

Evidence

result = await callables[tool_name](**arguments)

            return True, result

        except Exception as e:

            return False, str(e)

    async def shutdown(self) -> None:

        """

Remediation

Log the full exception server-side with a correlation ID; return only {"error_id": id, "message": "internal error"} to the caller. Never enable Flask debug mode in production.

Evidence

def run(cmd: list[str]) -> int:

    try:

        return subprocess.run(cmd, stderr=sys.stderr, stdout=sys.stderr).returncode

    except FileNotFoundError:

        return 127

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

script_path = shutil.which(script_name)

    if script_path:

        cmd = [script_path] + list(args[1:]) + ["--workspace", str(workspace)]

        return subprocess.run(cmd, check=False).returncode

    engine_dir = _REPO_ROOT / "engine"

    if (engine_dir / "antigravity_engine" / "__main__.py").exists():

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

**When** the agent attempts to execute:

```python

import subprocess

subprocess.run(["rm", "-rf", "/"])

```

**Then** execution is blocked before running  

**And** error message indicates blocked import: `subprocess`

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

"--workspace",

            str(workspace),

        ]

        return subprocess.run(cmd, cwd=str(engine_dir), check=False).returncode

    console.print(

        "[red]Engine not installed. Install: pip install ./engine "

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

# Build stage

FROM python:3.12-slim as builder

WORKDIR /app

COPY engine/ engine/

RUN pip install --user --no-cache-dir ./engine

# Runtime stage

FROM python:3.12-slim

WORKDIR /app

# Copy installed packages from builder

COPY --from=builder /root/.local /root/.local

ENV PATH=/root/.local/bin:$PATH

# Copy application code

COPY . .

# Set environment variables

ENV PYTHONUNBUFFERED=1

# Run the knowledge-hub MCP server against the mounted workspace

CMD ["ag-mcp", "--workspace", "/app"]

Remediation

Create and switch to a non-root user before the CMD / ENTRYPOINT: RUN adduser --system --uid 1000 app USER 1000 Or reuse the base image's shipped non-root account (e.g. `USER nobody`, `USER nonroot` on distroless). Multi-stage builds only need the USER directive in the final stage.

Evidence

- uses: actions/checkout@v3

    - name: Set up Python 3.12

      uses: actions/setup-python@v4

      with:

        python-version: "3.12"

Remediation

Pin every `uses:` to a 40-character commit SHA. Trailing comment with the version helps reviewers: `uses: actions/checkout@8e5e7e5ab8b370d6c329ec480221332ada57f0ab # v4.1.1` Automate the migration with `pinact` (https://github.com/suzuki-shunsuke/pinact) or `ratchet` (https://github.com/sethvargo/ratchet). Add a `pinact run --check` pre-commit hook so future PRs stay pinned. Re-pin when the action releases a new version — Dependabot can do this automatically with `version-update-strategy: inc

Evidence

runs-on: ubuntu-latest

    steps:

    - uses: actions/checkout@v3

    - name: Set up Python 3.12

      uses: actions/setup-python@v4

Remediation

Pin every `uses:` to a 40-character commit SHA. Trailing comment with the version helps reviewers: `uses: actions/checkout@8e5e7e5ab8b370d6c329ec480221332ada57f0ab # v4.1.1` Automate the migration with `pinact` (https://github.com/suzuki-shunsuke/pinact) or `ratchet` (https://github.com/sethvargo/ratchet). Add a `pinact run --check` pre-commit hook so future PRs stay pinned. Re-pin when the action releases a new version — Dependabot can do this automatically with `version-update-strategy: inc

Evidence

runs-on: ubuntu-latest

    steps:

    - uses: actions/checkout@v3

    - name: Validate docker-compose config

      run: docker compose config --quiet

Remediation

Pin every `uses:` to a 40-character commit SHA. Trailing comment with the version helps reviewers: `uses: actions/checkout@8e5e7e5ab8b370d6c329ec480221332ada57f0ab # v4.1.1` Automate the migration with `pinact` (https://github.com/suzuki-shunsuke/pinact) or `ratchet` (https://github.com/sethvargo/ratchet). Add a `pinact run --check` pre-commit hook so future PRs stay pinned. Re-pin when the action releases a new version — Dependabot can do this automatically with `version-update-strategy: inc

Evidence

runs-on: ubuntu-latest

    steps:

    - uses: actions/checkout@v3

    - name: Set up Python 3.12

      uses: actions/setup-python@v4

Remediation

Pin every `uses:` to a 40-character commit SHA. Trailing comment with the version helps reviewers: `uses: actions/checkout@8e5e7e5ab8b370d6c329ec480221332ada57f0ab # v4.1.1` Automate the migration with `pinact` (https://github.com/suzuki-shunsuke/pinact) or `ratchet` (https://github.com/sethvargo/ratchet). Add a `pinact run --check` pre-commit hook so future PRs stay pinned. Re-pin when the action releases a new version — Dependabot can do this automatically with `version-update-strategy: inc

Evidence

- uses: actions/checkout@v3

    - name: Set up Python 3.12

      uses: actions/setup-python@v4

      with:

        python-version: "3.12"

Remediation

Pin every `uses:` to a 40-character commit SHA. Trailing comment with the version helps reviewers: `uses: actions/checkout@8e5e7e5ab8b370d6c329ec480221332ada57f0ab # v4.1.1` Automate the migration with `pinact` (https://github.com/suzuki-shunsuke/pinact) or `ratchet` (https://github.com/sethvargo/ratchet). Add a `pinact run --check` pre-commit hook so future PRs stay pinned. Re-pin when the action releases a new version — Dependabot can do this automatically with `version-update-strategy: inc

Evidence

if relevant:

                lines.append("## Code Structure (relevant excerpts)\n")

                lines.append(relevant)

                lines.append("")

        except Exception:

            pass

    # -- Extract from knowledge_graph.md --

    kg = ag_dir / "knowledge_graph.md"

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

clean = _redact_secrets(message).replace("\n", " ").replace("\r", " ")

    fd = os.open(log_path, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600)

    try:

        os.chmod(log_path, 0o600)

    except OSError:

        pass

    with os.fdopen(fd, "a", encoding="utf-8") as handle:

        handle.write(f"{timestamp} {clean}\n")

    return log_path

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

(ag_dir / "module_registry.md").write_text(

                    _render_module_registry_markdown(registry_entries),

                    encoding="utf-8",

                )

            except Exception:

                pass  # Legacy registry is best-effort

        except Exception as exc:

            print(f"  ⚠ Map Agent failed: {exc}. Using fallback.", file=sys.stderr)

            # Fallback: build map.md from file listing

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

data = json.loads(pkg_text)

            main = data.get("main")

            if main:

                candidates.append(main)

        except Exception:

            pass

    candidates.extend(_COMMON_ENTRY_FILES)

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

structure = ""

    if structure_file.is_file():

        try:

            structure = structure_file.read_text(encoding="utf-8")

        except OSError:

            pass

    lines: list[str] = ["# Module Registry\n"]

    modules_dir = ag_dir / "modules"

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

try:

                content = mod_doc.read_text(encoding="utf-8")

                # Take first 800 chars — usually contains purpose + key files

                parts.append(f"**Deep knowledge (excerpt):**\n{content[:800]}")

            except OSError:

                pass

        # Source 2: structure.md section (AST-extracted)

        # Use resolve_module_path for accurate directory matching

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

conv_file = ag_dir / "conventions.md"

    if conv_file.is_file():

        try:

            conventions = conv_file.read_text(encoding="utf-8")[:2000]

        except OSError:

            pass

    prompt = f"""\

You are a senior software architect. Based on the evidence below, write an

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

(ag_dir / "module_registry.md").write_text(

                    _render_module_registry_markdown(registry_entries),

                    encoding="utf-8",

                )

            except Exception:

                pass

            _mark_stage_failure(

                refresh_status,

                stage="module_registry",

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

check=False,

        )

        if result.returncode == 0:

            return result.stdout.strip()

    except FileNotFoundError:

        pass

    return None

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

context: dict[str, str] = {}

    for md_file in sorted(ag_dir.glob("*.md")):

        try:

            context[md_file.name] = md_file.read_text(encoding="utf-8")

        except OSError:

            pass

    return context

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

if relevant:

                lines.append("## Knowledge Graph (relevant excerpts)\n")

                lines.append(relevant)

                lines.append("")

        except Exception:

            pass

    # -- Fallback: scan report summary --

    scan_report = ag_dir / "scan_report.json"

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

first_lines = agent_md.read_text(encoding="utf-8").splitlines()[:3]

                desc = " ".join(line.strip("#").strip() for line in first_lines if line.strip())

                if desc:

                    lines.append(f"**Description:** {desc[:200]}")

            except OSError:

                pass

        lines.append("")

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

binary = False

            try:

                with target.open("rb") as fh:

                    binary = b"\x00" in fh.read(2048)

            except OSError:

                pass

        except OSError as exc:

            return f"Error reading metadata '{file_path}': {exc}"

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

structure_file = ag_dir / "structure.md"

    if structure_file.is_file():

        try:

            structure = structure_file.read_text(encoding="utf-8")

        except OSError:

            pass

    module_evidence: list[str] = []

    for mod in modules:

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

doc_path = workspace / ".antigravity" / "module_registry.md"

    if doc_path.is_file():

        try:

            return doc_path.read_text(encoding="utf-8")

        except OSError:

            pass

    return None

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

doc_path = workspace / ".antigravity" / "modules" / "_git_insights.md"

    if doc_path.is_file():

        try:

            return doc_path.read_text(encoding="utf-8")

        except OSError:

            pass

    return "(No pre-generated git insights available. Use your tools to explore.)"

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

"[retrieval] "

                f"files={payload.get('file_count', 0)}, "

                f"elapsed={payload.get('scan_elapsed_seconds', 0.0)}s"

            )

        except Exception:

            pass

    if ("blueprint" in q.lower()) or ("模块" in q and "注册" in q):

        bps = _extract_blueprints_from_app(workspace)

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

continue

            # Fallback: treat entire top-level dir as one module

            modules.append(item.name)

    except OSError:

        pass

    if list_root_module_files(root):

        modules.append(WORKSPACE_ROOT_MODULE_ID)

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

doc_path = workspace / ".antigravity" / "map.md"

    if doc_path.is_file():

        try:

            return doc_path.read_text(encoding="utf-8")

        except OSError:

            pass

    return None

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

base = Path.home() / ".claude" / "plugins" / "data" / "antigravity-antigravity"

    base.mkdir(parents=True, exist_ok=True)

    try:

        os.chmod(base, 0o700)

    except OSError:

        pass

    return base / "ag-mcp.log"

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

scripts = data.get("project", {}).get("scripts", {})

                for _cmd, ref in scripts.items():

                    mod = ref.split(":")[0].replace(".", "/") + ".py"

                    candidates.append(mod)

            except Exception:

                pass

    pkg_text = config_contents.get("package.json", "")

    if pkg_text:

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

if total_bytes + len(text) > _CONFIG_TOTAL_LIMIT:

                    break

                contents[rel] = text

                total_bytes += len(text)

        except OSError:

            pass

    return contents

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

cleaned_count += 1

                    elif item.is_dir():

                        shutil.rmtree(item, ignore_errors=True)

                        cleaned_count += 1

                except Exception:

                    pass

    console.print(f"[green]✔ Cleaned {cleaned_count} temporary files/directories.[/green]")

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

doc_path = workspace / ".antigravity" / "structure.md"

    if doc_path.is_file():

        try:

            return doc_path.read_text(encoding="utf-8")

        except OSError:

            pass

    return "(No structure map available. Run `ag-refresh --workspace /path/to/project` first.)"

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

check=False,

        )

        if result.returncode == 0 and result.stdout.strip():

            parts.append("Contributors (last 3 months):\n" + result.stdout.strip())

    except FileNotFoundError:

        pass

    return "\n\n".join(parts)

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

if readme.exists():

            try:

                lines = readme.read_text(encoding="utf-8").splitlines()[:10]

                report.readme_snippet = "\n".join(lines)

            except OSError:

                pass

            break

    report.config_contents = _read_config_files(root)

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

files = list(dict.fromkeys(

                f.strip() for f in result.stdout.splitlines() if f.strip()

            ))[:20]

            parts.append("## Recently Modified Files\n" + "\n".join(f"- {f}" for f in files))

    except FileNotFoundError:

        pass

    try:

        result = subprocess.run(

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

capture_output=True, text=True, cwd=str(root), check=False,

        )

        if result.returncode == 0 and result.stdout.strip():

            parts.append("## Contributors (3 months)\n```\n" + result.stdout.strip() + "\n```")

    except FileNotFoundError:

        pass

    return "\n\n".join(parts)

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

try:

                    for sub in d.iterdir():

                        if sub.is_dir() and _is_venv_dir(sub):

                            venv_names.add(sub.name)

                except OSError:

                    pass

    return venv_names

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

"architecture", "design patterns", "file locations",

                            "key dependencies", "dependencies",

                        ):

                            tags.append(heading)

            except OSError:

                pass

        # Source 2: structure.md — extract key filenames

        if not tags:

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

legacy_path = ag_dir / "modules" / f"{module_name}.md"

    if legacy_path.is_file():

        try:

            return legacy_path.read_text(encoding="utf-8")

        except OSError:

            pass

    return "(No pre-generated knowledge available. Use your tools to explore.)"

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

if relevant:

                lines.append("## Project Conventions (relevant excerpts)\n")

                lines.append(relevant)

                lines.append("")

        except Exception:

            pass

    # -- Extract relevant sections from structure.md --

    structure = ag_dir / "structure.md"

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

agent_md = ag_dir / "agents" / f"{module_name}.md"

    if agent_md.is_file():

        try:

            return agent_md.read_text(encoding="utf-8")

        except OSError:

            pass

    # New format: agents/{module}/ (multi-group directory)

    agent_dir = ag_dir / "agents" / module_name

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

if child.name in skip:

                continue

            if _dir_has_code(child, venv_dirs):

                sub_modules.append(f"{parent_name}_{child.name}")

    except OSError:

        pass

    return sub_modules

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

and f.suffix.lower() in SOURCE_CODE_EXTS

                and "__pycache__" not in str(f)

            ):

                files.append(f"- {f.relative_to(mod_path)}")

    except OSError:

        pass

    return "\n".join(files[:50])

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

try:

            payload = json.loads(scan_report.read_text(encoding="utf-8"))

            file_count = int(payload.get("file_count", 0)) if isinstance(payload, dict) else 0

            lines.append(f"*(Project has {file_count} files. Try rephrasing with a more specific question.)*")

        except Exception:

            pass

    if len(lines) <= 4:

        lines.append("No relevant knowledge found. Try running `ag refresh` to rebuild the knowledge base.")

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.