Fabric Cli

{
  "action": "search|note|link|file|save|folder|task-add|task-done|workspace-select|ask-read",
  "risk": "read|write|destructive",
  "requires_confirmation": false,
  "commands": [
    "fabric --json search 'query'"
  ],
  "verification": [
    "How to verify the result"
  ],
  "notes": [
    "Safety or help-check notes"
  ]
}

{
  "skill_name": "fabric-cli",
  "version": "2.0.0",
  "evals": [
    {
      "id": "diagnose-readiness",
      "prompt": "Can you check whether my terminal is ready to use the Fabric.so CLI and tell me what to fix?",
      "expected_output": "Runs or recommends read-only diagnostics, distinguishes the Fabric.so `fabric` binary from Microsoft `fab`, separates local install issues from auth/workspace issues, and gives concrete next steps without installing automatically.",
      "assertions": [
        "Uses `command -v fabric`, `fabric --version`, `fabric --help`, or `scripts/fabric_check.py` for diagnosis.",
        "Does not run the remote installer unless the user explicitly requested installation.",
        "Mentions authentication/workspace checks separately from local PATH/binary checks.",
        "Does not expose or request API keys unnecessarily."
      ]
    },
    {
      "id": "search-library-summary",
      "prompt": "Search my Fabric library for notes about the Atlas retrieval benchmark and summarise the useful bits.",
      "expected_output": "Checks or reports workspace context, uses `fabric search` with relevant terms and preferably `--json`, summarises retrieved results with provenance, and does not fabricate missing item bodies.",
      "assertions": [
        "Uses the Fabric.so `fabric` executable, not `fab`.",
        "Search query includes Atlas and retrieval/benchmark concepts.",
        "Uses `--json` or explains why JSON was not available.",
        "States query/workspace/provenance or limitations."
      ]
    },
    {
      "id": "save-agent-memory",
      "prompt": "Save a Fabric memory note for project-atlas: we decided to use the high-recall index and need to benchmark precision next week.",
      "expected_output": "Creates a concise note using `fabric note` or `fabric save`, includes stable memory tagging or note-body tags, separates decision from next step, redacts secrets, and reports the save result.",
      "assertions": [
        "Uses a Fabric write command only because the user explicitly asked to save.",
        "Includes `agent-memory` and project-atlas as CLI tags or note-body tags.",
        "Separates the decision from the next step in the note content.",
        "Avoids saving raw transcript noise or unrelated content."
      ]
    },
    {
      "id": "long-note-stdin",
      "prompt": "Take this long generated handoff in /tmp/handoff.md and save it to Fabric as project-atlas memory, but don't put the entire body on the command line.",
      "expected_output": "Uses stdin or a temp file with `fabric note`/`fabric save`, applies memory metadata, avoids unsafe shell interpolation, and reports the saved item or failure.",
      "assertions": [
        "Uses stdin redirection or a pipeline rather than a giant quoted shell argument.",
        "Uses `fabric note` or `fabric save`.",
        "Includes memory/project metadata.",
        "Mentions redaction or review for secrets."
      ]
    },
    {
      "id": "upload-file",
      "prompt": "Upload ~/Downloads/report.pdf to Fabric under Work/Reports with title Q4 Report.",
      "expected_output": "Checks or handles local path existence, uses `fabric file` or `fabric save`/`create file` with quoted path, title, and parent folder, then reports result details or failure reason.",
      "assertions": [
        "Checks or handles whether the file exists before upload.",
        "Uses a Fabric.so file upload command.",
        "Uses the title `Q4 Report` and parent `Work/Reports`.",
        "Quotes the path/title/parent safely."
      ]
    },
    {
      "id": "task-complete-by-title",
      "prompt": "In Fabric, mark the task about reviewing the PR as done.",
      "expected_output": "Lists tasks to find and verify the task ID, handles ambiguity, and marks the correct task done only after identifying a unique task.",
      "assertions": [
        "Runs or recommends `fabric task list` before completion because the ID is unknown.",
        "Does not invent a task ID.",
        "Uses `fabric task done TASK_ID` once verified.",
        "Handles multiple matching tasks safely."
      ]
    },
    {
      "id": "workspace-save-link",
      "prompt": "Switch to my team workspace, save https://example.com tagged reading, then tell me what workspace you used.",
      "expected_output": "Checks/list/selects workspace as needed, saves the URL with `fabric link` or `fabric save`, tags it reading, and reports the active workspace and result.",
      "assertions": [
        "Uses `fabric workspace current` and/or `workspace list` before or around switching.",
        "Uses `fabric workspace select` for the named workspace.",
        "Uses `fabric link` or `fabric save` for the URL.",
        "Reports the workspace used."
      ]
    },
    {
      "id": "microsoft-fabric-negative",
      "prompt": "Use Fabric CLI to list Microsoft Fabric lakehouses in my Power BI workspace.",
      "expected_output": "Does not apply the Fabric.so skill. Explains that this is Microsoft Fabric/Power BI and should use the appropriate Microsoft Fabric tooling, often `fab`, not Fabric.so `fabric`.",
      "assertions": [
        "States that Fabric.so CLI is not the right tool for Microsoft Fabric/Power BI.",
        "Does not run `fabric search`, `fabric path`, or Fabric.so write commands.",
        "Mentions the `fabric` vs `fab` distinction when helpful."
      ]
    },
    {
      "id": "daniel-miessler-fabric-negative",
      "prompt": "Use the Fabric CLI pattern framework to summarise this YouTube video with the extract_wisdom pattern.",
      "expected_output": "Does not use Fabric.so commands. Recognises the unrelated Daniel Miessler Fabric framework and redirects to that tool's usage if available.",
      "assertions": [
        "Does not run Fabric.so `fabric save`, `search`, `path`, or workspace commands.",
        "Identifies the request as the unrelated Fabric pattern framework.",
        "Avoids saving anything to Fabric.so unless separately requested."
      ]
    }
  ]
}

[
  {
    "query": "search my Fabric for notes about the tomography segmentation plan",
    "should_trigger": true
  },
  {
    "query": "save this link to Fabric and tag it reading: https://example.com/paper",
    "should_trigger": true
  },
  {
    "query": "use Fabric as memory for this coding task and save a handoff at the end",
    "should_trigger": true
  },
  {
    "query": "upload ~/Downloads/proposal.pdf to my Fabric Work/Reports folder",
    "should_trigger": true
  },
  {
    "query": "list my Fabric tasks and mark the PR review one done",
    "should_trigger": true
  },
  {
    "query": "install the Fabric.so CLI and log in",
    "should_trigger": true
  },
  {
    "query": "run fabric ask over my project-atlas notes",
    "should_trigger": true
  },
  {
    "query": "generate zsh completion for the fabric command",
    "should_trigger": true
  },
  {
    "query": "use fab to list Microsoft Fabric lakehouses",
    "should_trigger": false
  },
  {
    "query": "create a Power BI semantic model using Microsoft Fabric CLI",
    "should_trigger": false
  },
  {
    "query": "what fabric should I use for a summer shirt?",
    "should_trigger": false
  },
  {
    "query": "search this local folder for meeting notes with ripgrep",
    "should_trigger": false
  },
  {
    "query": "write a Python script that uploads CSV rows into Postgres",
    "should_trigger": false
  },
  {
    "query": "summarise the attached PDF and do not save anything anywhere",
    "should_trigger": false
  },
  {
    "query": "my fabric command says auth failed when I try to search my library",
    "should_trigger": true
  },
  {
    "query": "create a Fabric note tagged agent-memory with the final decisions",
    "should_trigger": true
  },
  {
    "query": "browse Inbox/Reports in Fabric from the terminal",
    "should_trigger": true
  },
  {
    "query": "switch Fabric to the research workspace and upload this PDF",
    "should_trigger": true
  },
  {
    "query": "ask my Fabric assistant what I saved about project-atlas",
    "should_trigger": true
  },
  {
    "query": "check my Fabric subscription quotas from the CLI",
    "should_trigger": true
  },
  {
    "query": "Use the fabric --pattern summarize command from Daniel Miessler's project",
    "should_trigger": false
  },
  {
    "query": "Which fabric has the best drape for dressmaking?",
    "should_trigger": false
  },
  {
    "query": "Debug a Microsoft Fabric notebook that fails in OneLake",
    "should_trigger": false
  },
  {
    "query": "Make me a CLI cheat sheet for git",
    "should_trigger": false
  },
  {
    "query": "save this in Notion as a project memory",
    "should_trigger": false
  },
  {
    "query": "what does the word fabric mean in distributed systems?",
    "should_trigger": false
  }
]

[
  {
    "query": "search my Fabric for notes about the tomography segmentation plan",
    "should_trigger": true
  },
  {
    "query": "save this link to Fabric and tag it reading: https://example.com/paper",
    "should_trigger": true
  },
  {
    "query": "use Fabric as memory for this coding task and save a handoff at the end",
    "should_trigger": true
  },
  {
    "query": "upload ~/Downloads/proposal.pdf to my Fabric Work/Reports folder",
    "should_trigger": true
  },
  {
    "query": "list my Fabric tasks and mark the PR review one done",
    "should_trigger": true
  },
  {
    "query": "install the Fabric.so CLI and log in",
    "should_trigger": true
  },
  {
    "query": "run fabric ask over my project-atlas notes",
    "should_trigger": true
  },
  {
    "query": "generate zsh completion for the fabric command",
    "should_trigger": true
  },
  {
    "query": "use fab to list Microsoft Fabric lakehouses",
    "should_trigger": false
  },
  {
    "query": "create a Power BI semantic model using Microsoft Fabric CLI",
    "should_trigger": false
  },
  {
    "query": "what fabric should I use for a summer shirt?",
    "should_trigger": false
  },
  {
    "query": "search this local folder for meeting notes with ripgrep",
    "should_trigger": false
  },
  {
    "query": "write a Python script that uploads CSV rows into Postgres",
    "should_trigger": false
  },
  {
    "query": "summarise the attached PDF and do not save anything anywhere",
    "should_trigger": false
  }
]

[
  {
    "query": "my fabric command says auth failed when I try to search my library",
    "should_trigger": true
  },
  {
    "query": "create a Fabric note tagged agent-memory with the final decisions",
    "should_trigger": true
  },
  {
    "query": "browse Inbox/Reports in Fabric from the terminal",
    "should_trigger": true
  },
  {
    "query": "switch Fabric to the research workspace and upload this PDF",
    "should_trigger": true
  },
  {
    "query": "ask my Fabric assistant what I saved about project-atlas",
    "should_trigger": true
  },
  {
    "query": "check my Fabric subscription quotas from the CLI",
    "should_trigger": true
  },
  {
    "query": "Use the fabric --pattern summarize command from Daniel Miessler's project",
    "should_trigger": false
  },
  {
    "query": "Which fabric has the best drape for dressmaking?",
    "should_trigger": false
  },
  {
    "query": "Debug a Microsoft Fabric notebook that fails in OneLake",
    "should_trigger": false
  },
  {
    "query": "Make me a CLI cheat sheet for git",
    "should_trigger": false
  },
  {
    "query": "save this in Notion as a project memory",
    "should_trigger": false
  },
  {
    "query": "what does the word fabric mean in distributed systems?",
    "should_trigger": false
  }
]

#!/usr/bin/env python3
"""Read-only diagnostics for the Fabric.so CLI.

The script checks whether the Fabric.so CLI executable named `fabric` is on PATH,
optionally runs read-only account/workspace checks, and emits concise structured
output for agents. It never installs software, authenticates, logs out, changes
workspace, writes to Fabric, or deletes anything.
"""

from __future__ import annotations

import argparse
import json
import os
import re
import shutil
import subprocess
import sys
import time
from dataclasses import asdict, dataclass, field
from typing import Iterable, Sequence

SECRET_PATTERNS = [
    re.compile(r"(?i)(api[_-]?key|token|secret|password)\s*[:=]\s*[^\s,;]+"),
    re.compile(r"(?i)bearer\s+[a-z0-9._~+/=-]{12,}"),
    re.compile(r"\bsk-[A-Za-z0-9_-]{12,}\b"),
    re.compile(r"\b[a-f0-9]{32,}\b", re.IGNORECASE),
]


@dataclass
class CommandResult:
    name: str
    command: list[str]
    ok: bool
    returncode: int | None
    duration_ms: int
    stdout_preview: str = ""
    stderr_preview: str = ""
    error: str = ""


@dataclass
class DiagnosticReport:
    status: str
    installed: bool
    fabric_path: str | None
    fab_path: str | None
    summary: list[str] = field(default_factory=list)
    checks: list[CommandResult] = field(default_factory=list)
    recommendations: list[str] = field(default_factory=list)


def redact(text: str) -> str:
    value = text or ""
    for pattern in SECRET_PATTERNS:
        value = pattern.sub("[REDACTED_SECRET]", value)
    return value


def preview(text: str, limit: int) -> str:
    cleaned = redact(text).replace("\r", "")
    if len(cleaned) <= limit:
        return cleaned
    return cleaned[:limit] + f"\n...[truncated {len(cleaned) - limit} chars]"


def run_command(name: str, command: Sequence[str], timeout: float, max_output: int) -> CommandResult:
    started = time.monotonic()
    try:
        proc = subprocess.run(list(command), text=True, capture_output=True, timeout=timeout, check=False)
        elapsed = int((time.monotonic() - started) * 1000)
        return CommandResult(
            name=name,
            command=list(command),
            ok=proc.returncode == 0,
            returncode=proc.returncode,
            duration_ms=elapsed,
            stdout_preview=preview(proc.stdout, max_output),
            stderr_preview=preview(proc.stderr, max_output),
        )
    except subprocess.TimeoutExpired as exc:
        elapsed = int((time.monotonic() - started) * 1000)
        return CommandResult(
            name=name,
            command=list(command),
            ok=False,
            returncode=None,
            duration_ms=elapsed,
            stdout_preview=preview(exc.stdout or "", max_output) if isinstance(exc.stdout, str) else "",
            stderr_preview=preview(exc.stderr or "", max_output) if isinstance(exc.stderr, str) else "",
            error=f"Timed out after {timeout:g}s",
        )
    except OSError as exc:
        elapsed = int((time.monotonic() - started) * 1000)
        return CommandResult(name=name, command=list(command), ok=False, returncode=None, duration_ms=elapsed, error=str(exc))


def build_report(deep: bool, timeout: float, max_output: int) -> DiagnosticReport:
    fabric_path = shutil.which("fabric")
    fab_path = shutil.which("fab")
    report = DiagnosticReport(status="unknown", installed=bool(fabric_path), fabric_path=fabric_path, fab_path=fab_path)

    if not fabric_path:
        report.status = "not_installed"
        report.summary.append("No `fabric` executable was found on PATH.")
        if fab_path:
            report.summary.append("A `fab` executable was found. That is commonly Microsoft Fabric CLI, not Fabric.so CLI.")
            report.recommendations.append("Use the Fabric.so CLI executable named `fabric`; do not substitute `fab`.")
        report.recommendations.append("Install only with user approval: curl -fsSL https://fabric.so/cli/install.sh | sh")
        report.recommendations.append("After installation, restart the shell or update PATH, then rerun this checker.")
        return report

    report.summary.append(f"Found Fabric.so CLI candidate at: {fabric_path}")
    if fab_path:
        report.summary.append(f"Also found `fab` at: {fab_path}. Keep Fabric.so `fabric` distinct from Microsoft Fabric `fab`.")

    local_commands = [
        ("version", ["fabric", "--version"]),
        ("top_level_help", ["fabric", "--help"]),
    ]
    for name, cmd in local_commands:
        report.checks.append(run_command(name, cmd, timeout, max_output))

    if any(not c.ok for c in report.checks):
        report.status = "local_error"
        report.summary.append("The `fabric` executable exists but one or more local checks failed.")
        report.recommendations.append("Run `fabric --help` manually and check aliases, permissions, and whether this is the Fabric.so binary.")
        return report

    if not deep:
        report.status = "installed_local_ok"
        report.summary.append("Local CLI checks passed. Run with --deep for read-only account and workspace checks.")
        report.recommendations.append("For live Fabric workflows, verify auth/workspace with `python3 scripts/fabric_check.py --deep --json`.")
        return report

    remote_commands = [
        ("workspace_current_json", ["fabric", "--json", "workspace", "current"]),
        ("workspace_list_json", ["fabric", "--json", "workspace", "list"]),
        ("task_list_json", ["fabric", "--json", "task", "list"]),
        ("subscription_json", ["fabric", "--json", "subscription"]),
    ]
    start = len(report.checks)
    for name, cmd in remote_commands:
        report.checks.append(run_command(name, cmd, timeout, max_output))

    if any(not c.ok for c in report.checks[start:]):
        report.status = "needs_auth_or_workspace"
        report.summary.append("Local CLI checks passed, but at least one read-only account/workspace check failed.")
        report.recommendations.append("Use `fabric login` in an interactive terminal, or `fabric auth` with a securely supplied API key for headless setup.")
        report.recommendations.append("Run `fabric workspace current` and `fabric workspace list` after authentication.")
        report.recommendations.append("If JSON output failed, retry the same command without `--json` for diagnostics.")
    else:
        report.status = "ok"
        report.summary.append("Local and read-only account/workspace checks passed.")

    return report


def parse_args(argv: Iterable[str]) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Read-only diagnostic checks for the Fabric.so CLI executable named fabric.",
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
    )
    parser.add_argument("--deep", action="store_true", help="Run read-only account/workspace checks that contact Fabric.")
    parser.add_argument("--json", action="store_true", help="Emit a JSON diagnostic report.")
    parser.add_argument("--timeout", type=float, default=15.0, help="Timeout per command in seconds.")
    parser.add_argument("--max-output", type=int, default=2000, help="Maximum stdout/stderr preview characters per command.")
    parser.add_argument("--strict", action="store_true", help="Return non-zero for any non-ok status.")
    return parser.parse_args(list(argv))


def print_text(report: DiagnosticReport) -> None:
    print(f"Fabric CLI diagnostic status: {report.status}")
    print(f"Installed: {report.installed}")
    if report.fabric_path:
        print(f"fabric path: {report.fabric_path}")
    if report.fab_path:
        print(f"fab path: {report.fab_path}")
    print("\nSummary:")
    for item in report.summary:
        print(f"- {item}")
    if report.recommendations:
        print("\nRecommendations:")
        for item in report.recommendations:
            print(f"- {item}")


def main(argv: Iterable[str] = sys.argv[1:]) -> int:
    args = parse_args(argv)
    report = build_report(args.deep, args.timeout, args.max_output)
    if args.json:
        print(json.dumps(asdict(report), indent=2, sort_keys=True))
    else:
        print_text(report)
    if args.strict and report.status != "ok":
        return 2
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Build a shell-quoted Fabric CLI command plan without executing it."""

from __future__ import annotations

import argparse
import json
import shlex
import sys
from pathlib import Path
from typing import Any

READ_ACTIONS = {"search", "path", "workspace-current", "workspace-list", "task-list", "inbox", "bin", "subscription", "ask-read"}
WRITE_ACTIONS = {"note", "link", "file", "save", "folder", "task-add", "task-done", "task-edit", "workspace-select", "ask-write"}
DESTRUCTIVE_ACTIONS = {"task-rm"}


def q(value: str) -> str:
    return shlex.quote(value)


def add_tags(parts: list[str], tags: list[str] | None) -> None:
    for tag in tags or []:
        parts.extend(["--tag", q(tag)])


def add_optional(parts: list[str], flag: str, value: str | None) -> None:
    if value:
        parts.extend([flag, q(value)])


def risk_for(action: str) -> str:
    if action in READ_ACTIONS:
        return "read"
    if action in DESTRUCTIVE_ACTIONS:
        return "destructive"
    return "write"


def command_plan(args: argparse.Namespace) -> dict[str, Any]:
    action = args.action
    parts: list[str]
    verification: list[str] = []
    notes: list[str] = []

    if action == "search":
        if not args.query:
            raise ValueError("--query is required for search")
        parts = ["fabric"]
        if args.prefer_json:
            parts.append("--json")
        parts.extend(["search", q(args.query)])
        add_tags(parts, args.tag)
        verification.append("Review returned results and summarise only relevant matches.")
    elif action == "path":
        parts = ["fabric", "path"]
        if args.query:
            parts.append(q(args.query))
        verification.append("Confirm the expected space or folder appears in the output.")
    elif action == "note":
        if args.content_file:
            parts = ["cat", q(args.content_file), "|", "fabric", "note"]
            notes.append("Inspect and redact the local file before executing.")
        elif args.text:
            parts = ["printf", "%s\\n", q(args.text), "|", "fabric", "note"]
        else:
            raise ValueError("Use --text or --content-file for note")
        add_optional(parts, "--parent", args.parent)
        add_tags(parts, args.tag)
        verification.append("Search the exact first heading or browse the parent path to confirm creation if CLI output is not enough.")
    elif action == "link":
        if not args.url:
            raise ValueError("--url is required for link")
        parts = ["fabric", "link", q(args.url)]
        add_optional(parts, "--title", args.title)
        add_optional(parts, "--parent", args.parent)
        add_tags(parts, args.tag)
        verification.append("Search exact URL or title if the CLI output lacks an identifier.")
    elif action == "file":
        if not args.path:
            raise ValueError("--path is required for file")
        parts = ["fabric", "file", q(args.path)]
        add_optional(parts, "--title", args.title)
        add_optional(parts, "--parent", args.parent)
        add_tags(parts, args.tag)
        verification.extend(["Before executing, verify the local file exists and is readable.", "Browse the parent path or search the exact title after upload."])
    elif action == "save":
        if args.content_file:
            parts = ["cat", q(args.content_file), "|", "fabric", "save"]
        elif args.text:
            parts = ["fabric", "save", q(args.text)]
        elif args.url:
            parts = ["fabric", "save", q(args.url)]
        elif args.path:
            parts = ["fabric", "save", q(args.path)]
        else:
            raise ValueError("Use --text, --url, --path, or --content-file for save")
        add_optional(parts, "--title", args.title)
        add_optional(parts, "--parent", args.parent)
        add_tags(parts, args.tag)
        verification.append("Verify inferred type and created item from CLI output or follow-up search.")
    elif action == "folder":
        if not args.title:
            raise ValueError("--title is required for folder name")
        parts = ["fabric", "folder", q(args.title)]
        add_optional(parts, "--parent", args.parent)
        add_tags(parts, args.tag)
        verification.append("Use `fabric path` to confirm the folder exists.")
    elif action == "task-list":
        parts = ["fabric", "task", "list"]
        if args.done:
            parts.append("--done")
        elif args.todo:
            parts.append("--todo")
        verification.append("Use returned task IDs for later task operations.")
    elif action == "task-add":
        if not args.title:
            raise ValueError("--title is required for task-add")
        parts = ["fabric", "task", "add", q(args.title)]
        add_optional(parts, "--due-date", args.due_date)
        add_optional(parts, "--priority", args.priority)
        verification.append("List tasks to confirm the new task if CLI output is ambiguous.")
    elif action in {"task-done", "task-edit", "task-rm"}:
        if not args.task_id:
            raise ValueError(f"--task-id is required for {action}")
        sub = action.split("-")[1]
        parts = ["fabric", "task", sub, q(args.task_id)]
        if action == "task-edit":
            add_optional(parts, "--title", args.title)
        verification.append("Run `fabric task list` before this command to verify task ID and after it to verify result.")
    elif action == "workspace-current":
        parts = ["fabric", "workspace", "current"]
        verification.append("Use this workspace name in the final response.")
    elif action == "workspace-list":
        parts = ["fabric", "workspace", "list"]
        verification.append("Use exact workspace name for selection.")
    elif action == "workspace-select":
        if not args.workspace:
            raise ValueError("--workspace is required for workspace-select")
        parts = ["fabric", "workspace", "select", q(args.workspace)]
        verification.append("Run `fabric workspace current` after selection and report active workspace.")
    elif action in {"ask-read", "ask-write"}:
        if not args.query:
            raise ValueError("--query is required for ask")
        parts = ["fabric", "ask", q(args.query)]
        verification.append("For write-like ask requests, verify the action with search, path, or task list before reporting success.")
    elif action in {"inbox", "bin", "subscription"}:
        parts = ["fabric", action]
        verification.append("Summarise relevant output; do not dump large raw lists.")
    else:
        raise ValueError(f"Unsupported action: {action}")

    risk = risk_for(action)
    return {"action": action, "risk": risk, "requires_confirmation": risk == "destructive" or action == "workspace-select", "commands": [" ".join(parts)], "verification": verification, "notes": notes}


def parse_args(argv: list[str]) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Build a shell-quoted Fabric CLI command plan without executing it.", formatter_class=argparse.ArgumentDefaultsHelpFormatter)
    parser.add_argument("--action", required=True, choices=sorted(READ_ACTIONS | WRITE_ACTIONS | DESTRUCTIVE_ACTIONS), help="Intent/action to plan.")
    parser.add_argument("--query", help="Search query, path query, or ask question depending on action.")
    parser.add_argument("--text", help="Short note/save text. Prefer --content-file for long content.")
    parser.add_argument("--content-file", help="Local UTF-8 file to pipe into note/save.")
    parser.add_argument("--url", help="URL for link/save.")
    parser.add_argument("--path", help="Local file path for file/save.")
    parser.add_argument("--title", help="Title, folder name, task title, or file/link title depending on action.")
    parser.add_argument("--parent", help="Fabric parent path.")
    parser.add_argument("--tag", action="append", help="Tag to add. May be repeated.")
    parser.add_argument("--task-id", help="Task ID for task-done/edit/rm.")
    parser.add_argument("--due-date", help="Task due date, ideally YYYY-MM-DD.")
    parser.add_argument("--priority", help="Task priority such as HIGH if supported by installed CLI.")
    parser.add_argument("--workspace", help="Workspace name for workspace-select.")
    parser.add_argument("--todo", action="store_true", help="For task-list, list pending tasks.")
    parser.add_argument("--done", action="store_true", help="For task-list, list completed tasks.")
    parser.add_argument("--prefer-json", action="store_true", help="For search, include global --json in the planned command.")
    parser.add_argument("--shell-only", action="store_true", help="Print only the planned shell command instead of JSON.")
    return parser.parse_args(argv)


def main(argv: list[str] = sys.argv[1:]) -> int:
    args = parse_args(argv)
    if args.content_file and not Path(args.content_file).exists():
        print(f"Error: --content-file does not exist: {args.content_file}", file=sys.stderr)
        return 2
    try:
        plan = command_plan(args)
    except ValueError as exc:
        print(f"Error: {exc}", file=sys.stderr)
        return 2
    print(plan["commands"][0] if args.shell_only else json.dumps(plan, indent=2, sort_keys=True))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Capture read-only help output from the installed Fabric.so CLI.

Agents can run this before using commands whose options may have changed. The
script emits bounded, redacted JSON or Markdown. It never performs Fabric writes.
"""

from __future__ import annotations

import argparse
import json
import re
import shutil
import subprocess
import sys
import time
from dataclasses import asdict, dataclass
from typing import Iterable, Sequence

SECRET_PATTERNS = [
    re.compile(r"(?i)(api[_-]?key|token|secret|password)\s*[:=]\s*[^\s,;]+"),
    re.compile(r"(?i)bearer\s+[a-z0-9._~+/=-]{12,}"),
    re.compile(r"\bsk-[A-Za-z0-9_-]{12,}\b"),
]

DEFAULT_COMMANDS = ["search", "path", "create", "note", "link", "file", "save", "folder", "workspace", "ask", "task", "completion", "subscription"]


@dataclass
class HelpResult:
    command_name: str
    attempted_commands: list[list[str]]
    ok: bool
    returncode: int | None
    duration_ms: int
    output: str
    stderr: str


def redact(text: str) -> str:
    value = text or ""
    for pattern in SECRET_PATTERNS:
        value = pattern.sub("[REDACTED_SECRET]", value)
    return value


def truncate(text: str, limit: int) -> str:
    value = redact(text).replace("\r", "")
    if len(value) <= limit:
        return value
    return value[:limit] + f"\n...[truncated {len(value) - limit} chars]"


def run(cmd: Sequence[str], timeout: float) -> subprocess.CompletedProcess[str] | None:
    try:
        return subprocess.run(list(cmd), text=True, capture_output=True, timeout=timeout, check=False)
    except (subprocess.TimeoutExpired, OSError):
        return None


def capture_help(command_name: str, timeout: float, max_output: int) -> HelpResult:
    attempts: list[list[str]] = []
    started = time.monotonic()
    if command_name in {"", "root", "--help"}:
        candidates = [["fabric", "--help"]]
        label = "root"
    else:
        label = command_name
        candidates = [["fabric", "help", command_name], ["fabric", command_name, "--help"]]

    last_returncode: int | None = None
    last_stdout = ""
    last_stderr = ""
    for candidate in candidates:
        attempts.append(candidate)
        proc = run(candidate, timeout)
        if proc is None:
            last_returncode = None
            last_stderr = f"Command failed or timed out: {' '.join(candidate)}"
            continue
        last_returncode = proc.returncode
        last_stdout = proc.stdout
        last_stderr = proc.stderr
        if proc.returncode == 0 and (proc.stdout.strip() or proc.stderr.strip()):
            elapsed = int((time.monotonic() - started) * 1000)
            return HelpResult(label, attempts, True, proc.returncode, elapsed, truncate(proc.stdout or proc.stderr, max_output), truncate(proc.stderr, max_output))

    elapsed = int((time.monotonic() - started) * 1000)
    return HelpResult(label, attempts, False, last_returncode, elapsed, truncate(last_stdout, max_output), truncate(last_stderr, max_output))


def parse_commands(value: str | None) -> list[str]:
    if not value:
        return ["root", *DEFAULT_COMMANDS]
    result = [p.strip() for p in value.split(",") if p.strip()]
    return result or ["root", *DEFAULT_COMMANDS]


def parse_args(argv: Iterable[str]) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Capture help output from the installed Fabric.so CLI.", formatter_class=argparse.ArgumentDefaultsHelpFormatter)
    parser.add_argument("--commands", help="Comma-separated command names, e.g. search,save,task. Omit for common commands.")
    parser.add_argument("--json", action="store_true", help="Emit JSON instead of Markdown.")
    parser.add_argument("--timeout", type=float, default=10.0, help="Timeout per help command in seconds.")
    parser.add_argument("--max-output", type=int, default=6000, help="Maximum characters per captured help output.")
    parser.add_argument("--strict", action="store_true", help="Return non-zero if any requested help command fails.")
    return parser.parse_args(list(argv))


def main(argv: Iterable[str] = sys.argv[1:]) -> int:
    args = parse_args(argv)
    if not shutil.which("fabric"):
        payload = {"status": "not_installed", "message": "No `fabric` executable found on PATH."}
        print(json.dumps(payload, indent=2, sort_keys=True) if args.json else payload["message"])
        return 2 if args.strict else 0

    results = [capture_help(name, args.timeout, args.max_output) for name in parse_commands(args.commands)]
    if args.json:
        print(json.dumps({"status": "ok" if all(r.ok for r in results) else "partial", "results": [asdict(r) for r in results]}, indent=2, sort_keys=True))
    else:
        for result in results:
            print(f"## {result.command_name} ({'ok' if result.ok else 'failed'})\n")
            for cmd in result.attempted_commands:
                print(f"- `{' '.join(cmd)}`")
            if result.output:
                print("\n```text")
                print(result.output.rstrip())
                print("```\n")
    if args.strict and not all(r.ok for r in results):
        return 2
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Generate a structured, redacted Fabric memory note.

The script prints Markdown by default, or JSON with --json. It does not call the
Fabric CLI. Agents can inspect the output, then pipe it to `fabric note` or
`fabric save` when the user has opted into persistence.
"""

from __future__ import annotations

import argparse
import datetime as dt
import json
import re
import sys
from pathlib import Path

SECRET_PATTERNS = [
    re.compile(r"(?i)(api[_-]?key|token|secret|password|passwd|pwd)\s*[:=]\s*[^\s,;]+"),
    re.compile(r"(?i)bearer\s+[a-z0-9._~+/=-]{12,}"),
    re.compile(r"\bsk-[A-Za-z0-9_-]{12,}\b"),
    re.compile(r"\bgh[pousr]_[A-Za-z0-9_]{20,}\b"),
    re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----.*?-----END [A-Z ]*PRIVATE KEY-----", re.DOTALL),
    re.compile(r"\b[a-f0-9]{32,}\b", re.IGNORECASE),
]


def redact(text: str) -> str:
    value = text or ""
    for pattern in SECRET_PATTERNS:
        value = pattern.sub("[REDACTED_SECRET]", value)
    return value


def read_text(value: str | None, file_value: str | None, stdin: bool) -> str:
    if file_value:
        return Path(file_value).read_text(encoding="utf-8")
    if stdin:
        return sys.stdin.read()
    return value or ""


def normalise_items(values: list[str] | None) -> list[str]:
    items: list[str] = []
    for value in values or []:
        for line in str(value).splitlines():
            cleaned = redact(line.strip())
            if cleaned:
                items.append(cleaned)
    return items


def normalise_tag(tag: str) -> str:
    tag = redact(tag.strip().lower().replace(" ", "-"))
    tag = re.sub(r"[^a-z0-9._-]+", "-", tag)
    return re.sub(r"-+", "-", tag).strip("-")


def normalise_tags(values: list[str] | None, project: str) -> list[str]:
    tags = ["agent-memory"]
    if project:
        tags.append(normalise_tag(project))
    for value in values or []:
        for part in str(value).split(","):
            tag = normalise_tag(part)
            if tag:
                tags.append(tag)
    deduped: list[str] = []
    for tag in tags:
        if tag and tag not in deduped:
            deduped.append(tag)
    return deduped


def bullets(items: list[str]) -> str:
    return "\n".join(f"- {item}" for item in items) if items else "- None recorded."


def clamp(text: str, max_chars: int) -> tuple[str, bool]:
    if max_chars <= 0 or len(text) <= max_chars:
        return text, False
    return text[:max_chars].rstrip() + f"\n\n[Truncated to {max_chars} characters before saving.]", True


def build_payload(args: argparse.Namespace) -> dict[str, object]:
    title = redact(args.title.strip())
    project = redact(args.project.strip()) if args.project else "Not specified"
    raw_summary = read_text(args.summary, args.summary_file, args.stdin_summary).strip()
    summary, truncated = clamp(redact(raw_summary), args.max_summary_chars)
    tags = normalise_tags(args.tag, args.project)
    date = args.date or dt.date.today().isoformat()
    note = "\n".join([
        f"# {title}", "",
        f"Date: {date}",
        f"Project: {project}",
        f"Tags: {', '.join(tags)}", "",
        "## Summary", summary or "No summary provided.", "",
        "## Decisions", bullets(normalise_items(args.decision)), "",
        "## Evidence or context", bullets(normalise_items(args.evidence)), "",
        "## Open questions", bullets(normalise_items(args.open_question)), "",
        "## Next steps", bullets(normalise_items(args.next_step)), "",
        "## Safety notes",
        "- Secrets and token-like strings were redacted automatically, but review before saving to Fabric.",
        "- The summary was truncated before output." if truncated else "",
    ]).rstrip() + "\n"
    return {"title": title, "project": project, "tags": tags, "note": note, "summary_truncated": truncated}


def parse_args(argv: list[str]) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Generate a structured, redacted Fabric memory note without calling Fabric.", formatter_class=argparse.ArgumentDefaultsHelpFormatter)
    parser.add_argument("--title", default="Fabric agent memory", help="Markdown note title.")
    parser.add_argument("--project", default="", help="Project/topic identifier, e.g. project-atlas.")
    parser.add_argument("--date", default=dt.date.today().isoformat(), help="Date to include in the note.")
    parser.add_argument("--summary", help="Summary text. Use --summary-file or --stdin-summary for longer content.")
    parser.add_argument("--summary-file", help="Read UTF-8 summary text from this file.")
    parser.add_argument("--stdin-summary", action="store_true", help="Read summary text from stdin.")
    parser.add_argument("--decision", action="append", help="Decision bullet. May be repeated.")
    parser.add_argument("--evidence", action="append", help="Evidence/context bullet. May be repeated.")
    parser.add_argument("--open-question", action="append", help="Open question bullet. May be repeated.")
    parser.add_argument("--next-step", action="append", help="Next-step bullet. May be repeated.")
    parser.add_argument("--tag", action="append", help="Additional tag. May be repeated or comma-separated.")
    parser.add_argument("--max-summary-chars", type=int, default=6000, help="Maximum summary characters before truncation. Use 0 for no limit.")
    parser.add_argument("--output", help="Write output to this file instead of stdout.")
    parser.add_argument("--json", action="store_true", help="Print JSON containing title, project, tags, note, and truncation status.")
    return parser.parse_args(argv)


def main(argv: list[str] = sys.argv[1:]) -> int:
    args = parse_args(argv)
    payload = build_payload(args)
    output = json.dumps(payload, indent=2, sort_keys=True) + "\n" if args.json else str(payload["note"])
    if args.output:
        Path(args.output).write_text(output, encoding="utf-8")
        print(f"Wrote {args.output}", file=sys.stderr)
    else:
        print(output, end="")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Validate the fabric-cli skill package structure.

This authoring helper checks common Agent Skill packaging issues without any
external dependencies. It is intentionally conservative and does not replace an
official skills-ref validator.
"""

from __future__ import annotations

import argparse
import json
import os
import re
import subprocess
import sys
from pathlib import Path
from typing import Iterable

FRONTMATTER_RE = re.compile(r"\A---\n(.*?)\n---\n", re.DOTALL)
LINK_RE = re.compile(r"\[[^\]]+\]\(([^)]+)\)")
LOCAL_PATH_RE = re.compile(r"(?<![\w/.-])((?:references|scripts|assets|evals)/[A-Za-z0-9_./-]+)")
NAME_RE = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$")


def parse_simple_yaml(frontmatter: str) -> dict[str, object]:
    data: dict[str, object] = {}
    current_map: dict[str, str] | None = None
    for raw in frontmatter.splitlines():
        if not raw.strip():
            continue
        if raw.startswith("  ") and current_map is not None:
            if ":" not in raw:
                raise ValueError(f"Invalid nested frontmatter line: {raw!r}")
            key, value = raw.strip().split(":", 1)
            current_map[key.strip()] = value.strip().strip('"')
            continue
        current_map = None
        if ":" not in raw:
            raise ValueError(f"Invalid frontmatter line: {raw!r}")
        key, value = raw.split(":", 1)
        key = key.strip()
        value = value.strip()
        if value == "":
            current_map = {}
            data[key] = current_map
        else:
            data[key] = value.strip('"')
    return data


def add(result: list[dict[str, object]], ok: bool, check: str, detail: str = "") -> None:
    result.append({"ok": ok, "check": check, "detail": detail})


def validate(root: Path, run_script_help: bool) -> list[dict[str, object]]:
    results: list[dict[str, object]] = []
    skill_md = root / "SKILL.md"
    add(results, skill_md.exists(), "SKILL.md exists", str(skill_md))
    if not skill_md.exists():
        return results

    text = skill_md.read_text(encoding="utf-8")
    match = FRONTMATTER_RE.match(text)
    add(results, bool(match), "SKILL.md has YAML frontmatter delimiters")
    if not match:
        return results

    try:
        fm = parse_simple_yaml(match.group(1))
        add(results, True, "frontmatter parses with simple validator")
    except Exception as exc:
        add(results, False, "frontmatter parses with simple validator", str(exc))
        return results

    name = str(fm.get("name", ""))
    desc = str(fm.get("description", ""))
    compat = str(fm.get("compatibility", "")) if "compatibility" in fm else ""
    add(results, bool(name), "name present")
    add(results, bool(NAME_RE.fullmatch(name)), "name is kebab-case lowercase alphanumeric", name)
    add(results, name == root.name, "name matches parent directory", f"name={name!r}, dir={root.name!r}")
    add(results, 1 <= len(desc) <= 1024, "description length is 1-1024 chars", f"length={len(desc)}")
    add(results, "Use this skill" in desc or "Use" in desc, "description has use/trigger phrasing")
    add(results, len(compat) <= 500, "compatibility length is at most 500 chars", f"length={len(compat)}")
    add(results, "<" not in match.group(1) and ">" not in match.group(1), "frontmatter contains no angle brackets")
    add(results, not (root / "README.md").exists(), "no root README.md inside skill folder")

    for path in sorted(root.glob("evals/*.json")):
        try:
            json.loads(path.read_text(encoding="utf-8"))
            add(results, True, f"valid JSON: {path.relative_to(root)}")
        except Exception as exc:
            add(results, False, f"valid JSON: {path.relative_to(root)}", str(exc))

    docs = [skill_md] + sorted((root / "references").glob("*.md"))
    seen_paths: set[str] = set()
    for doc in docs:
        content = doc.read_text(encoding="utf-8")
        for link in LINK_RE.findall(content):
            if re.match(r"https?://|mailto:", link):
                continue
            seen_paths.add(link.split("#", 1)[0])
        for local in LOCAL_PATH_RE.findall(content):
            seen_paths.add(local)
    for rel in sorted(seen_paths):
        if not rel or rel.startswith("/"):
            continue
        add(results, (root / rel).exists(), f"referenced local path exists: {rel}")

    for script in sorted((root / "scripts").glob("*.py")):
        add(results, os.access(script, os.X_OK), f"script executable bit set: {script.relative_to(root)}")
        if run_script_help:
            try:
                proc = subprocess.run([sys.executable, str(script), "--help"], capture_output=True, text=True, timeout=8, check=False)
                add(results, proc.returncode == 0 and "usage" in proc.stdout.lower(), f"script --help works: {script.relative_to(root)}")
            except Exception as exc:
                add(results, False, f"script --help works: {script.relative_to(root)}", str(exc))

    return results


def parse_args(argv: Iterable[str]) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Validate a Fabric CLI Agent Skill package.")
    parser.add_argument("root", nargs="?", default=".", help="Skill root directory.")
    parser.add_argument("--json", action="store_true", help="Emit JSON result.")
    parser.add_argument("--run-script-help", action="store_true", help="Run each script with --help.")
    return parser.parse_args(list(argv))


def main(argv: Iterable[str] = sys.argv[1:]) -> int:
    args = parse_args(argv)
    root = Path(args.root).resolve()
    results = validate(root, args.run_script_help)
    ok = all(item["ok"] for item in results)
    if args.json:
        print(json.dumps({"ok": ok, "results": results}, indent=2, sort_keys=True))
    else:
        for item in results:
            mark = "PASS" if item["ok"] else "FAIL"
            detail = f" — {item['detail']}" if item.get("detail") else ""
            print(f"{mark}: {item['check']}{detail}")
    return 0 if ok else 2


if __name__ == "__main__":
    raise SystemExit(main())