feat: add Tigris CLI harness for S3-compatible object storage (#301)

* add tigris data harness

* use Tigris CLI

* Fix Tigris harness safety and packaging

---------

Co-authored-by: yuhao <itsyuhao@icloud.com>

.gitignore

@@ -100,6 +100,7 @@
 !/3MF/
 !/calibre/
 !/rekordbox/
+!/tigris/
 !/cc-switch/
 !/siyuan/
 # Step 5: Inside each software dir, ignore everything (including dotfiles)
@@ -209,6 +210,8 @@
 /calibre/.*
 /rekordbox/*
 /rekordbox/.*
+/tigris/*
+/tigris/.*
 /cc-switch/*
 /cc-switch/.*
 /siyuan/*
@@ -278,6 +281,7 @@
 !/quietshrink/agent-harness/
 !/mailchimp/agent-harness/
 !/rekordbox/agent-harness/
+!/tigris/agent-harness/
 !/cc-switch/agent-harness/
 !/siyuan/agent-harness/
 

README.md

@@ -1306,6 +1306,13 @@ Each application received complete, production-ready CLI interfaces — not demo
 <td align="center">✅ 50</td>
 </tr>
 <tr>
+<td align="center"><strong>🗄️ <a href="tigris/agent-harness/">Tigris</a></strong></td>
+<td>Object Storage (S3-compatible, global, no egress) — buckets, objects, snapshots, IAM, access keys</td>
+<td><code>cli-anything-tigris</code></td>
+<td>subprocess wrapping the official <code>tigris</code> CLI</td>
+<td align="center">✅ <a href="tigris/agent-harness/">New</a></td>
+</tr>
+<tr>
 <td align="center" colspan="4"><strong>Total</strong></td>
 <td align="center"><strong>✅ 2,461</strong></td>
 </tr>

registry.json

@@ -1342,6 +1342,25 @@
           "url": "https://github.com/hito0512"
         }
       ]
+    },
+    {
+      "name": "tigris",
+      "display_name": "Tigris",
+      "version": "1.0.0",
+      "description": "Object storage management — buckets, objects, presigned URLs, snapshots, IAM, scoped access keys. Wraps the official `tigris` CLI (S3-compatible, globally distributed, no egress fees)",
+      "requires": "Tigris CLI on PATH (`npm install -g @tigrisdata/cli` or `brew install tigrisdata/tap/tigris`) + `tigris login`",
+      "homepage": "https://www.tigrisdata.com",
+      "source_url": null,
+      "install_cmd": "pip install git+https://github.com/HKUDS/CLI-Anything.git#subdirectory=tigris/agent-harness",
+      "entry_point": "cli-anything-tigris",
+      "skill_md": "skills/cli-anything-tigris/SKILL.md",
+      "category": "storage",
+      "contributors": [
+        {
+          "name": "davidmyriel",
+          "url": "https://github.com/davidmyriel"
+        }
+      ]
     }
   ]
 }

skills/cli-anything-tigris/SKILL.md

@@ -0,0 +1,199 @@
+---
+name: "cli-anything-tigris"
+description: >-
+  Command-line interface for Tigris object storage — wraps the official `tigris` CLI to expose buckets, objects, presigned URLs, snapshots, IAM, and scoped access keys to AI agents. Globally distributed, S3-compatible, no egress fees.
+---
+
+# cli-anything-tigris
+
+A stateless command-line interface for [Tigris](https://www.tigrisdata.com)
+object storage. Wraps the official `tigris` CLI so every Tigris primitive
+(snapshots, IAM, scoped credentials, OAuth) is reachable through a single
+agent-friendly entry point with `--json` everywhere.
+
+## Scope
+
+This harness is for the official Tigris CLI only. It shells out to the
+`tigris` binary and assumes Tigris auth through `tigris login` or Tigris
+access keys. It is not a generic S3 endpoint tool, and it does not manage
+MinIO, Cloudflare R2, AWS S3, or arbitrary S3-compatible endpoints.
+
+## Installation
+
+```bash
+# 1. Install the underlying Tigris CLI
+npm install -g @tigrisdata/cli
+# or:
+brew install tigrisdata/tap/tigris
+
+# 2. Authenticate (browser OAuth)
+tigris login
+
+# 3. Install this harness
+pip install cli-anything-tigris
+```
+
+**Prerequisites:**
+- Python 3.10+
+- `tigris` CLI on PATH (the binary's alias is `t3`)
+
+## Usage
+
+### Basic Commands
+
+```bash
+# Show help
+cli-anything-tigris --help
+
+# Start interactive REPL
+cli-anything-tigris
+
+# Whoami (JSON output for agents)
+cli-anything-tigris --json auth whoami
+
+# List buckets
+cli-anything-tigris --json bucket list
+
+# Upload a local file
+cli-anything-tigris --json object put --bucket my-bucket --key path/to/file.txt --file ./local.txt
+
+# Download an object
+cli-anything-tigris --json object get --bucket my-bucket --key path/to/file.txt --output ./out.txt
+
+# Server-side copy
+cli-anything-tigris --json object cp t3://my-bucket/src.txt t3://my-bucket/dst.txt
+
+# Take a snapshot
+cli-anything-tigris --json snapshot take my-bucket --name baseline-v1
+
+# Delete a bucket; --yes is required
+cli-anything-tigris --json bucket delete --name old-bucket --yes
+
+# Create a scoped access key for an agent run
+cli-anything-tigris --json access-key create my-agent-key
+cli-anything-tigris --json access-key assign tid_AaBb --bucket my-bucket --role Editor
+
+# Rotate or delete access keys; --yes is required
+cli-anything-tigris --json access-key rotate tid_AaBb --yes
+cli-anything-tigris --json access-key delete tid_AaBb --yes
+
+# Presigned download URL (1 hour)
+cli-anything-tigris --json presign get --bucket my-bucket --key path/to/file.txt --expires 3600
+```
+
+### REPL Mode
+
+When invoked without a subcommand, the CLI enters an interactive REPL with
+tab-completion and history.
+
+## Command Groups
+
+### auth
+OAuth-based authentication.
+
+| Command | Description |
+|---------|-------------|
+| `login` | Browser OAuth login (`tigris login`) |
+| `logout` | Log out of current session |
+| `whoami` | Print authenticated user / org |
+
+### bucket
+Bucket CRUD.
+
+| Command | Description |
+|---------|-------------|
+| `list` | List all buckets |
+| `create --name NAME` | Create a bucket |
+| `delete --name NAME --yes` | Delete an empty bucket (`--yes` required) |
+| `info NAME` | Get bucket info |
+
+### object
+Object operations (wraps `tigris ls/cp/rm/stat`).
+
+| Command | Description |
+|---------|-------------|
+| `list --bucket B [--prefix P] [--limit N]` | List objects |
+| `put --bucket B --key K (--file F \| --text T)` | Upload an object |
+| `get --bucket B --key K --output F` | Download an object |
+| `delete --bucket B --key K` | Delete an object |
+| `info --bucket B --key K` | Object metadata (HEAD / stat) |
+| `cp SRC DST [-r]` | Copy. Accepts `t3://` or `tigris://` URIs. |
+
+### presign
+Time-limited URLs.
+
+| Command | Description |
+|---------|-------------|
+| `get --bucket B --key K [--expires SEC]` | Presigned download URL |
+| `put --bucket B --key K [--expires SEC]` | Presigned upload URL |
+
+### snapshot
+Point-in-time bucket snapshots — a Tigris-specific primitive.
+
+| Command | Description |
+|---------|-------------|
+| `list BUCKET` | List snapshots for a bucket |
+| `take BUCKET [--name N]` | Take a snapshot |
+
+### access-key
+Scoped programmatic credentials — combine with `snapshot` for per-agent isolation.
+
+| Command | Description |
+|---------|-------------|
+| `list` | List all access keys |
+| `create NAME` | Create a new access key (secret shown ONCE) |
+| `get KEY_ID` | Show key details |
+| `delete KEY_ID --yes` | Permanently delete a key (`--yes` required) |
+| `assign KEY_ID --bucket B --role R` | Scope a key to a bucket + role |
+| `rotate KEY_ID --yes` | Rotate a key's secret (`--yes` required) |
+
+### iam
+Policies and organization users.
+
+| Command | Description |
+|---------|-------------|
+| `policy list` | List IAM policies |
+| `policy create NAME --document FILE` | Create a policy from a JSON file |
+| `user list` | List org users |
+| `user invite EMAIL [--role R]` | Invite a user |
+
+## Output Formats
+
+All commands support dual output modes:
+
+- **Human-readable** (default): tables, colors, formatted text
+- **Machine-readable** (`--json`): JSON envelope (or upstream CLI's
+  `--format json` output, echoed verbatim)
+
+## For AI Agents
+
+When using this CLI programmatically:
+
+1. Always pass `--json` for parseable output.
+2. Check return codes — 0 for success, non-zero for errors.
+3. Read stderr for error messages.
+4. `object cp` accepts `t3://bucket/key` or `tigris://bucket/key` URIs;
+   server-side copies (t3 → t3) skip the round trip entirely.
+5. `presign` returns a URL on stdout in human mode; in JSON mode it's the
+   `url` field.
+6. **For destructive work**: bucket deletion, access-key deletion, and
+   access-key rotation require explicit `--yes`; take a `snapshot` of the
+   target bucket first, then do the work, then either keep the snapshot or
+   discard.
+7. **For per-agent isolation**: `access-key create` + `access-key assign
+   --bucket B --role Editor` to mint a key scoped to one bucket; revoke with
+   `access-key delete` when the agent run ends.
+
+## Why Tigris
+
+- **Globally distributed.** Data placed close to wherever it's read.
+- **No egress fees.** Agents pulling artifacts from anywhere don't incur
+  per-region bandwidth charges.
+- **Snapshots + scoped credentials.** Primitives generic S3-compatible
+  providers don't ship — the foundation for per-agent isolation.
+- **S3-compatible.** Useful alongside S3-aware tools, but this harness
+  itself is not a generic S3/MinIO/R2/AWS endpoint manager.
+
+## Version
+
+1.0.0

tigris/agent-harness/.gitignore

@@ -0,0 +1,28 @@
+# Python
+__pycache__/
+*.py[cod]
+*$py.class
+*.so
+
+# Package build
+*.egg-info/
+dist/
+build/
+*.egg
+
+# Testing
+.pytest_cache/
+.coverage
+htmlcov/
+
+# IDE
+.idea/
+.vscode/
+*.swp
+
+# Local virtual envs
+.venv/
+venv/
+
+# Session files
+.cli-anything-tigris/

tigris/agent-harness/TIGRIS.md

@@ -0,0 +1,124 @@
+# Agent Harness: Tigris Object Storage CLI
+
+## Purpose
+
+This harness provides a standard operating procedure (SOP) and toolkit for
+coding agents (Claude Code, Codex, etc.) to interact with
+[Tigris](https://www.tigrisdata.com) — a globally distributed, S3-compatible
+object storage service with no egress fees.
+
+The harness **wraps the official `tigris` CLI** rather than reimplementing
+the S3 protocol. This means agents get access to every Tigris primitive
+(snapshots, IAM, scoped access keys, organizations, OAuth) — not just
+generic S3 ops — and the harness inherits new commands as the upstream CLI
+ships them.
+
+This is intentionally Tigris CLI-only tooling. Do not use this harness as a
+generic S3 endpoint wrapper, and do not treat it as MinIO, Cloudflare R2, AWS
+S3, or arbitrary S3-compatible endpoint management. It shells out to the
+official `tigris` binary and follows that CLI's auth and command model.
+
+## Requirements
+
+- **Python 3.10+** (uses PEP 604 union syntax and PEP 585 generic types).
+  On macOS the system Python is 3.9 — use `pyenv`, `uv`, or
+  `brew install python@3.12`.
+- **The Tigris CLI on PATH.** Install with one of:
+
+  ```bash
+  npm install -g @tigrisdata/cli
+  brew install tigrisdata/tap/tigris
+  ```
+
+  Then `tigris login` once to authenticate (browser OAuth).
+
+## Backend Description
+
+Each command in this harness builds a `tigris <args> --format json`
+invocation, runs it via `subprocess.run`, and parses the JSON output.
+Commands the upstream CLI does not JSON-format (e.g. `login`, `cp`) are
+streamed directly to the caller's TTY or captured as text.
+
+Credentials are normally resolved by the `tigris` CLI itself (via the OAuth
+session created by `tigris login`). Explicit `--access-key` / `--secret-key`
+flags export `TIGRIS_STORAGE_*` and `AWS_*` env vars into the child process
+for setups that rely on env-based auth.
+
+## Architecture
+
+```
+agent-harness/
+├── .gitignore
+├── setup.py                           # cli, prompt-toolkit; no boto3
+├── TIGRIS.md                          # this file
+└── cli_anything/
+    └── tigris/
+        ├── __init__.py
+        ├── __main__.py                # python -m entry point
+        ├── README.md                  # usage docs
+        ├── tigris_cli.py              # click CLI + REPL dispatcher
+        ├── core/
+        │   ├── auth.py                # login, logout, whoami
+        │   ├── bucket.py              # list, create, delete, info
+        │   ├── object.py              # list, put, get, delete, info, cp
+        │   ├── presign.py             # presign get/put
+        │   ├── snapshot.py            # list, take
+        │   ├── access_key.py          # list, create, get, delete, assign, rotate
+        │   └── iam.py                 # policies + users
+        ├── utils/
+        │   ├── tigris_backend.py      # subprocess wrapper around `tigris`
+        │   └── repl_skin.py           # unified REPL skin (unmodified copy)
+        ├── skills/
+        │   └── SKILL.md
+        └── tests/
+            ├── TEST.md
+            ├── test_core.py           # subprocess.run mocked
+            └── test_full_e2e.py       # real `tigris` CLI on PATH, env-gated
+```
+
+## Command Groups
+
+| Group        | What it wraps                          | Operations |
+|--------------|----------------------------------------|------------|
+| `auth`       | `tigris login/logout/whoami`           | login, logout, whoami |
+| `bucket`     | `tigris buckets ...`                   | list, create, delete --yes, info |
+| `object`     | `tigris ls/cp/rm/stat`                 | list, put, get, delete, info, cp |
+| `presign`    | `tigris presign`                       | get, put |
+| `snapshot`   | `tigris snapshots ...`                 | list, take |
+| `access-key` | `tigris access-keys ...`               | list, create, get, delete --yes, assign, rotate --yes |
+| `iam`        | `tigris iam policies / users ...`      | policy list/create, user list/invite |
+
+## Output Modes
+
+- **Human-readable** (default): tables, colors, formatted text via the REPL skin
+- **Machine-readable** (`--json`): the upstream CLI's `--format json` output,
+  echoed verbatim (with a thin envelope for commands that don't natively
+  support JSON, like `cp`).
+
+## Agent Usage
+
+When agents drive this CLI:
+
+1. Pass `--json` for parseable output.
+2. Inspect return codes (0 = success).
+3. Read stderr for error messages.
+4. Use `object cp t3://src/key t3://dst/key` for server-side copies — no data
+   flows through the agent, no egress charges.
+5. Use `presign get/put` to hand off object access to other tools or
+   downstream agents without sharing credentials.
+6. Use `snapshot take` before destructive work to make a recovery point;
+   `snapshot list` to find one to restore from.
+7. Use `access-key create` + `access-key assign --bucket B --role Editor` to
+   mint a scoped key for an agent run, then `access-key delete` to revoke.
+8. Bucket deletion, access-key deletion, and access-key rotation require
+   explicit `--yes`; without it, the harness refuses to call the backend.
+
+## Testing
+
+- `tests/test_core.py` — unit tests with `subprocess.run` fully mocked;
+  passable without the `tigris` CLI installed or any network access.
+- `tests/test_full_e2e.py` — real-bucket tests; gated on
+  `CLI_ANYTHING_TIGRIS_RUN_E2E=1` plus `tigris` on PATH and an authenticated
+  session.
+
+See `tests/TEST.md` for run instructions.

tigris/agent-harness/cli_anything/tigris/README.md

@@ -0,0 +1,46 @@
+# cli-anything-tigris
+
+CLI-Anything harness for [Tigris](https://www.tigrisdata.com) — a globally
+distributed, S3-compatible object storage service with no egress fees.
+
+This harness **wraps the official `tigris` CLI**, so every Tigris primitive
+(snapshots, IAM, scoped access keys, OAuth login) is reachable through one
+agent-friendly entry point with `--json` everywhere.
+
+It is Tigris CLI-only tooling, not a generic S3/MinIO/R2/AWS endpoint
+manager.
+
+## Install
+
+```bash
+# 1. Install the underlying Tigris CLI
+npm install -g @tigrisdata/cli      # or: brew install tigrisdata/tap/tigris
+
+# 2. Authenticate (browser OAuth)
+tigris login
+
+# 3. Install this harness
+pip install cli-anything-tigris
+```
+
+## Quick start
+
+```bash
+# Interactive REPL
+cli-anything-tigris
+
+# Or drive directly
+cli-anything-tigris --json auth whoami
+cli-anything-tigris --json bucket list
+cli-anything-tigris --json object cp ./local.bin t3://my-bucket/remote.bin
+cli-anything-tigris --json snapshot take my-bucket --name baseline-v1
+cli-anything-tigris --json access-key create agent-run-42
+cli-anything-tigris --json access-key rotate tid_AaBb --yes
+cli-anything-tigris --json presign get --bucket my-bucket --key hello.txt
+```
+
+Bucket deletion, access-key deletion, and access-key rotation require
+explicit `--yes`.
+
+See [SKILL.md](skills/SKILL.md) for the full command reference and
+agent-usage guidance.

tigris/agent-harness/cli_anything/tigris/__main__.py

@@ -0,0 +1,6 @@
+"""Entry point for `python -m cli_anything.tigris`."""
+
+from .tigris_cli import main
+
+if __name__ == "__main__":
+    main()

tigris/agent-harness/cli_anything/tigris/core/access_key.py

@@ -0,0 +1,188 @@
+"""Access key commands -- list, create, get, delete, assign, rotate.
+
+Wraps `tigris access-keys`. Scoped per-bucket roles are one of Tigris's
+agent-storage primitives — combine with `snapshot` and per-agent buckets to
+give each agent its own least-privilege credentials.
+"""
+
+import json as json_mod
+import click
+
+from ..utils.tigris_backend import TigrisBackend, TigrisCliError
+
+
+@click.group("access-key")
+@click.pass_context
+def access_key_group(ctx):
+    """Manage Tigris access keys (wraps `tigris access-keys`)."""
+    pass
+
+
+@access_key_group.command("list")
+@click.pass_context
+def list_keys(ctx):
+    """List all access keys in the current organization."""
+    backend: TigrisBackend = ctx.obj["backend"]
+    use_json = ctx.obj.get("json", False)
+    skin = ctx.obj.get("skin")
+    try:
+        keys = backend.list_access_keys()
+        if use_json:
+            click.echo(json_mod.dumps(keys, indent=2))
+        else:
+            if not keys:
+                skin.info("No access keys found.")
+                return
+            if isinstance(keys, list):
+                headers = ["ID", "Name", "Created"]
+                rows = []
+                for k in keys:
+                    if not isinstance(k, dict):
+                        rows.append([str(k), "", ""])
+                        continue
+                    rows.append([
+                        str(k.get("id") or k.get("accessKeyId") or "?"),
+                        str(k.get("name") or ""),
+                        str(k.get("created") or k.get("createdAt") or ""),
+                    ])
+                skin.table(headers, rows)
+            else:
+                click.echo(keys)
+    except TigrisCliError as e:
+        _emit_error(use_json, skin, f"Failed to list access keys: {e}")
+        raise SystemExit(1)
+
+
+@access_key_group.command("create")
+@click.argument("name")
+@click.pass_context
+def create_key(ctx, name):
+    """Create a new access key. Secret is shown only once."""
+    backend: TigrisBackend = ctx.obj["backend"]
+    use_json = ctx.obj.get("json", False)
+    skin = ctx.obj.get("skin")
+    try:
+        result = backend.create_access_key(name)
+        if use_json:
+            click.echo(json_mod.dumps(result, indent=2))
+        else:
+            skin.success(f"Access key '{name}' created — secret shown ONCE, save it now")
+            if isinstance(result, dict):
+                for k, v in result.items():
+                    skin.status(k, str(v))
+            else:
+                click.echo(result)
+    except TigrisCliError as e:
+        _emit_error(use_json, skin, f"Failed to create access key: {e}")
+        raise SystemExit(1)
+
+
+@access_key_group.command("get")
+@click.argument("key_id")
+@click.pass_context
+def get_key(ctx, key_id):
+    """Show details for an access key."""
+    backend: TigrisBackend = ctx.obj["backend"]
+    use_json = ctx.obj.get("json", False)
+    skin = ctx.obj.get("skin")
+    try:
+        info = backend.get_access_key(key_id)
+        if use_json:
+            click.echo(json_mod.dumps(info, indent=2))
+        else:
+            skin.section(f"Access key: {key_id}")
+            if isinstance(info, dict):
+                for k, v in info.items():
+                    skin.status(k, str(v))
+            else:
+                click.echo(info)
+    except TigrisCliError as e:
+        _emit_error(use_json, skin, f"Failed to get access key: {e}")
+        raise SystemExit(1)
+
+
+@access_key_group.command("delete")
+@click.argument("key_id")
+@click.option("--yes", is_flag=True, default=False,
+              help="Required. Confirm access-key deletion and pass --yes to Tigris.")
+@click.pass_context
+def delete_key(ctx, key_id, yes):
+    """Permanently delete an access key."""
+    backend: TigrisBackend = ctx.obj["backend"]
+    use_json = ctx.obj.get("json", False)
+    skin = ctx.obj.get("skin")
+    if not yes:
+        _emit_error(use_json, skin, "Refusing to delete access key without --yes")
+        raise SystemExit(1)
+    try:
+        result = backend.delete_access_key(key_id, yes=yes)
+        if use_json:
+            click.echo(json_mod.dumps(result or {"id": key_id, "status": "deleted"}, indent=2))
+        else:
+            skin.success(f"Access key {key_id} deleted")
+    except TigrisCliError as e:
+        _emit_error(use_json, skin, f"Failed to delete access key: {e}")
+        raise SystemExit(1)
+
+
+@access_key_group.command("assign")
+@click.argument("key_id")
+@click.option("--bucket", required=True, help="Bucket to scope the key to")
+@click.option("--role", required=True,
+              help="Role to grant (e.g. Editor, Viewer)")
+@click.pass_context
+def assign_key(ctx, key_id, bucket, role):
+    """Assign a per-bucket role to an access key (scoped credentials)."""
+    backend: TigrisBackend = ctx.obj["backend"]
+    use_json = ctx.obj.get("json", False)
+    skin = ctx.obj.get("skin")
+    try:
+        result = backend.assign_access_key(key_id, bucket=bucket, role=role)
+        if use_json:
+            click.echo(json_mod.dumps(
+                result or {"id": key_id, "bucket": bucket, "role": role,
+                           "status": "assigned"}, indent=2,
+            ))
+        else:
+            skin.success(f"Key {key_id} assigned {role} on {bucket}")
+    except TigrisCliError as e:
+        _emit_error(use_json, skin, f"Failed to assign access key: {e}")
+        raise SystemExit(1)
+
+
+@access_key_group.command("rotate")
+@click.argument("key_id")
+@click.option("--yes", is_flag=True, default=False,
+              help="Required. Confirm secret rotation and pass --yes to Tigris.")
+@click.pass_context
+def rotate_key(ctx, key_id, yes):
+    """Rotate an access key's secret."""
+    backend: TigrisBackend = ctx.obj["backend"]
+    use_json = ctx.obj.get("json", False)
+    skin = ctx.obj.get("skin")
+    if not yes:
+        _emit_error(use_json, skin, "Refusing to rotate access key without --yes")
+        raise SystemExit(1)
+    try:
+        result = backend.rotate_access_key(key_id, yes=yes)
+        if use_json:
+            click.echo(json_mod.dumps(result or {"id": key_id, "status": "rotated"}, indent=2))
+        else:
+            skin.success(f"Access key {key_id} rotated — new secret shown ONCE")
+            if isinstance(result, dict):
+                for k, v in result.items():
+                    skin.status(k, str(v))
+            else:
+                click.echo(result)
+    except TigrisCliError as e:
+        _emit_error(use_json, skin, f"Failed to rotate access key: {e}")
+        raise SystemExit(1)
+
+
+def _emit_error(use_json: bool, skin, message: str) -> None:
+    if use_json:
+        click.echo(json_mod.dumps({"error": message}, indent=2))
+    elif skin:
+        skin.error(message)
+    else:
+        click.echo(message, err=True)

tigris/agent-harness/cli_anything/tigris/core/auth.py

@@ -0,0 +1,74 @@
+"""Auth commands -- login, logout, whoami.
+
+Wraps `tigris login`, `tigris logout`, `tigris whoami`.
+"""
+
+import json as json_mod
+import click
+
+from ..utils.tigris_backend import TigrisBackend, TigrisCliError
+
+
+@click.group("auth")
+@click.pass_context
+def auth_group(ctx):
+    """Authentication (wraps `tigris login/logout/whoami`)."""
+    pass
+
+
+@auth_group.command("login")
+@click.pass_context
+def login(ctx):
+    """Interactive OAuth login. Streams `tigris login` to your terminal."""
+    backend: TigrisBackend = ctx.obj["backend"]
+    skin = ctx.obj.get("skin")
+    try:
+        backend.login()
+        if skin:
+            skin.success("Logged in.")
+    except TigrisCliError as e:
+        if skin:
+            skin.error(f"Login failed: {e}")
+        raise SystemExit(1)
+
+
+@auth_group.command("logout")
+@click.pass_context
+def logout(ctx):
+    """Log out of the current Tigris session."""
+    backend: TigrisBackend = ctx.obj["backend"]
+    skin = ctx.obj.get("skin")
+    try:
+        backend.logout()
+        if skin:
+            skin.success("Logged out.")
+    except TigrisCliError as e:
+        if skin:
+            skin.error(f"Logout failed: {e}")
+        raise SystemExit(1)
+
+
+@auth_group.command("whoami")
+@click.pass_context
+def whoami(ctx):
+    """Print the currently authenticated user / organization."""
+    backend: TigrisBackend = ctx.obj["backend"]
+    use_json = ctx.obj.get("json", False)
+    skin = ctx.obj.get("skin")
+    try:
+        info = backend.whoami()
+        if use_json:
+            click.echo(json_mod.dumps(info, indent=2))
+        else:
+            skin.section("Authentication")
+            if isinstance(info, dict):
+                for k, v in info.items():
+                    skin.status(k, str(v))
+            else:
+                click.echo(info)
+    except TigrisCliError as e:
+        if use_json:
+            click.echo(json_mod.dumps({"error": str(e)}, indent=2))
+        else:
+            skin.error(f"whoami failed: {e}")
+        raise SystemExit(1)

tigris/agent-harness/cli_anything/tigris/core/bucket.py

@@ -0,0 +1,120 @@
+"""Bucket commands -- list, create, delete, info."""
+
+import json as json_mod
+import click
+
+from ..utils.tigris_backend import TigrisBackend, TigrisCliError
+
+
+@click.group("bucket")
+@click.pass_context
+def bucket_group(ctx):
+    """Manage Tigris buckets (wraps `tigris buckets`)."""
+    pass
+
+
+@bucket_group.command("list")
+@click.pass_context
+def list_buckets(ctx):
+    """List all buckets in the current organization."""
+    backend: TigrisBackend = ctx.obj["backend"]
+    use_json = ctx.obj.get("json", False)
+    skin = ctx.obj.get("skin")
+    try:
+        buckets = backend.list_buckets()
+        if use_json:
+            click.echo(json_mod.dumps(buckets, indent=2))
+        else:
+            if not buckets:
+                skin.info("No buckets found.")
+                return
+            if isinstance(buckets, list):
+                # Normalize to a couple of expected fields if present.
+                rows = []
+                headers = ["Name", "Created"]
+                for b in buckets:
+                    name = b.get("name") or b.get("Name") or "?"
+                    created = b.get("created") or b.get("CreationDate") or ""
+                    rows.append([name, str(created)])
+                skin.table(headers, rows)
+            else:
+                click.echo(buckets)
+    except TigrisCliError as e:
+        _emit_error(use_json, skin, f"Failed to list buckets: {e}")
+        raise SystemExit(1)
+
+
+@bucket_group.command("create")
+@click.option("--name", required=True, help="Bucket name to create")
+@click.pass_context
+def create_bucket(ctx, name):
+    """Create a new bucket."""
+    backend: TigrisBackend = ctx.obj["backend"]
+    use_json = ctx.obj.get("json", False)
+    skin = ctx.obj.get("skin")
+    try:
+        result = backend.create_bucket(name)
+        if use_json:
+            click.echo(json_mod.dumps(result or {"name": name, "status": "created"}, indent=2))
+        else:
+            skin.success(f"Bucket '{name}' created")
+    except TigrisCliError as e:
+        _emit_error(use_json, skin, f"Failed to create bucket: {e}")
+        raise SystemExit(1)
+
+
+@bucket_group.command("delete")
+@click.option("--name", required=True, help="Bucket name to delete")
+@click.option("--yes", is_flag=True, default=False,
+              help="Required. Confirm bucket deletion and pass --yes to Tigris.")
+@click.pass_context
+def delete_bucket(ctx, name, yes):
+    """Delete an empty bucket."""
+    backend: TigrisBackend = ctx.obj["backend"]
+    use_json = ctx.obj.get("json", False)
+    skin = ctx.obj.get("skin")
+    if not yes:
+        _emit_error(use_json, skin, "Refusing to delete bucket without --yes")
+        raise SystemExit(1)
+    try:
+        result = backend.delete_bucket(name, yes=yes)
+        if use_json:
+            click.echo(json_mod.dumps(result or {"name": name, "status": "deleted"}, indent=2))
+        else:
+            skin.success(f"Bucket '{name}' deleted")
+    except TigrisCliError as e:
+        _emit_error(use_json, skin, f"Failed to delete bucket: {e}")
+        raise SystemExit(1)
+
+
+@bucket_group.command("info")
+@click.argument("name")
+@click.pass_context
+def bucket_info(ctx, name):
+    """Get info about a bucket (`tigris buckets get`)."""
+    backend: TigrisBackend = ctx.obj["backend"]
+    use_json = ctx.obj.get("json", False)
+    skin = ctx.obj.get("skin")
+    try:
+        info = backend.head_bucket(name)
+        if use_json:
+            click.echo(json_mod.dumps(info or {"name": name}, indent=2))
+        else:
+            skin.section(f"Bucket: {name}")
+            if isinstance(info, dict):
+                for k, v in info.items():
+                    skin.status(k, str(v))
+            else:
+                click.echo(info)
+    except TigrisCliError as e:
+        _emit_error(use_json, skin, f"Failed to get bucket info: {e}")
+        raise SystemExit(1)
+
+
+def _emit_error(use_json: bool, skin, message: str) -> None:
+    if use_json:
+        click.echo(json_mod.dumps({"error": message}, indent=2))
+    elif skin:
+        skin.error(message)
+    else:
+        click.echo(message, err=True)

tigris/agent-harness/cli_anything/tigris/core/iam.py

@@ -0,0 +1,162 @@
+"""IAM commands -- policies and users.
+
+Wraps `tigris iam policies` and `tigris iam users`. Together with
+`access-key assign`, IAM policies let agents/orgs grant narrow, auditable
+permissions to programmatic clients.
+"""
+
+import json as json_mod
+import click
+
+from ..utils.tigris_backend import TigrisBackend, TigrisCliError
+
+
+@click.group("iam")
+@click.pass_context
+def iam_group(ctx):
+    """IAM — policies and users (wraps `tigris iam`)."""
+    pass
+
+
+# ── policies ──────────────────────────────────────────────────────────
+
+
+@iam_group.group("policy")
+@click.pass_context
+def policy_group(ctx):
+    """Manage IAM policies."""
+    pass
+
+
+@policy_group.command("list")
+@click.pass_context
+def list_policies(ctx):
+    """List all IAM policies."""
+    backend: TigrisBackend = ctx.obj["backend"]
+    use_json = ctx.obj.get("json", False)
+    skin = ctx.obj.get("skin")
+    try:
+        policies = backend.list_iam_policies()
+        if use_json:
+            click.echo(json_mod.dumps(policies, indent=2))
+        else:
+            if not policies:
+                skin.info("No IAM policies found.")
+                return
+            if isinstance(policies, list):
+                headers = ["Name", "ARN / ID", "Created"]
+                rows = []
+                for p in policies:
+                    if not isinstance(p, dict):
+                        rows.append([str(p), "", ""])
+                        continue
+                    rows.append([
+                        str(p.get("name") or "?"),
+                        str(p.get("arn") or p.get("id") or ""),
+                        str(p.get("created") or p.get("createdAt") or ""),
+                    ])
+                skin.table(headers, rows)
+            else:
+                click.echo(policies)
+    except TigrisCliError as e:
+        _emit_error(use_json, skin, f"Failed to list policies: {e}")
+        raise SystemExit(1)
+
+
+@policy_group.command("create")
+@click.argument("name")
+@click.option("--document", required=True,
+              help="Path to a JSON policy document")
+@click.pass_context
+def create_policy(ctx, name, document):
+    """Create a new IAM policy from a JSON document."""
+    backend: TigrisBackend = ctx.obj["backend"]
+    use_json = ctx.obj.get("json", False)
+    skin = ctx.obj.get("skin")
+    try:
+        result = backend.create_iam_policy(name, document)
+        if use_json:
+            click.echo(json_mod.dumps(result or {"name": name, "status": "created"}, indent=2))
+        else:
+            skin.success(f"Policy '{name}' created from {document}")
+    except TigrisCliError as e:
+        _emit_error(use_json, skin, f"Failed to create policy: {e}")
+        raise SystemExit(1)
+
+
+# ── users ─────────────────────────────────────────────────────────────
+
+
+@iam_group.group("user")
+@click.pass_context
+def user_group(ctx):
+    """Manage organization users and invitations."""
+    pass
+
+
+@user_group.command("list")
+@click.pass_context
+def list_users(ctx):
+    """List all users in the current organization."""
+    backend: TigrisBackend = ctx.obj["backend"]
+    use_json = ctx.obj.get("json", False)
+    skin = ctx.obj.get("skin")
+    try:
+        users = backend.list_iam_users()
+        if use_json:
+            click.echo(json_mod.dumps(users, indent=2))
+        else:
+            if not users:
+                skin.info("No users found.")
+                return
+            if isinstance(users, list):
+                headers = ["Email", "Role", "Status"]
+                rows = []
+                for u in users:
+                    if not isinstance(u, dict):
+                        rows.append([str(u), "", ""])
+                        continue
+                    rows.append([
+                        str(u.get("email") or "?"),
+                        str(u.get("role") or ""),
+                        str(u.get("status") or ""),
+                    ])
+                skin.table(headers, rows)
+            else:
+                click.echo(users)
+    except TigrisCliError as e:
+        _emit_error(use_json, skin, f"Failed to list users: {e}")
+        raise SystemExit(1)
+
+
+@user_group.command("invite")
+@click.argument("email")
+@click.option("--role", default="member",
+              help="Role to assign on join (default: member)")
+@click.pass_context
+def invite_user(ctx, email, role):
+    """Invite a user by email."""
+    backend: TigrisBackend = ctx.obj["backend"]
+    use_json = ctx.obj.get("json", False)
+    skin = ctx.obj.get("skin")
+    try:
+        result = backend.invite_iam_user(email, role=role)
+        if use_json:
+            click.echo(json_mod.dumps(
+                result or {"email": email, "role": role, "status": "invited"},
+                indent=2,
+            ))
+        else:
+            skin.success(f"Invited {email} as {role}")
+    except TigrisCliError as e:
+        _emit_error(use_json, skin, f"Failed to invite user: {e}")
+        raise SystemExit(1)
+
+
+def _emit_error(use_json: bool, skin, message: str) -> None:
+    if use_json:
+        click.echo(json_mod.dumps({"error": message}, indent=2))
+    elif skin:
+        skin.error(message)
+    else:
+        click.echo(message, err=True)

tigris/agent-harness/cli_anything/tigris/core/object.py

@@ -0,0 +1,220 @@
+"""Object commands -- list, put, get, delete, info, cp."""
+
+import json as json_mod
+import click
+
+from ..utils.tigris_backend import TigrisBackend, TigrisCliError
+
+
+T3_SCHEMES = ("t3://", "tigris://")
+
+
+def _is_remote(path: str) -> bool:
+    return path.startswith(T3_SCHEMES)
+
+
+def _parse_tigris_uri(uri: str) -> tuple[str, str]:
+    """Parse t3://bucket/key or tigris://bucket/key into (bucket, key)."""
+    for scheme in T3_SCHEMES:
+        if uri.startswith(scheme):
+            rest = uri[len(scheme):]
+            parts = rest.split("/", 1)
+            if len(parts) == 2 and parts[0] and parts[1]:
+                return parts[0], parts[1]
+            raise click.UsageError(
+                f"Expected {scheme}<bucket>/<key>, got: {uri}"
+            )
+    raise click.UsageError(
+        f"Expected URI starting with t3:// or tigris://, got: {uri}"
+    )
+
+
+@click.group("object")
+@click.pass_context
+def object_group(ctx):
+    """Manage Tigris objects (wraps `tigris ls/cp/rm/stat`)."""
+    pass
+
+
+@object_group.command("list")
+@click.option("--bucket", required=True, help="Bucket name")
+@click.option("--prefix", default=None, help="Filter objects by key prefix")
+@click.option("--limit", default=None, type=int,
+              help="Limit to first N results (client-side trim)")
+@click.pass_context
+def list_objects(ctx, bucket, prefix, limit):
+    """List objects in a bucket (`tigris ls`)."""
+    backend: TigrisBackend = ctx.obj["backend"]
+    use_json = ctx.obj.get("json", False)
+    skin = ctx.obj.get("skin")
+    try:
+        objs = backend.list_objects(bucket, prefix=prefix, limit=limit)
+        if use_json:
+            click.echo(json_mod.dumps(objs, indent=2))
+        else:
+            if not objs:
+                skin.info(f"No objects found in '{bucket}'.")
+                return
+            if isinstance(objs, list):
+                headers = ["Key", "Size", "Modified"]
+                rows = []
+                for o in objs:
+                    if not isinstance(o, dict):
+                        rows.append([str(o), "", ""])
+                        continue
+                    rows.append([
+                        str(o.get("key") or o.get("Key") or "?"),
+                        str(o.get("size") or o.get("Size") or ""),
+                        str(o.get("modified") or o.get("LastModified") or ""),
+                    ])
+                skin.table(headers, rows)
+            else:
+                click.echo(objs)
+    except TigrisCliError as e:
+        _emit_error(use_json, skin, f"Failed to list objects: {e}")
+        raise SystemExit(1)
+
+
+@object_group.command("put")
+@click.option("--bucket", required=True, help="Bucket name")
+@click.option("--key", required=True, help="Object key")
+@click.option("--file", "file_path", default=None,
+              help="Local file path to upload (uses `tigris cp`)")
+@click.option("--text", default=None,
+              help="Inline text content (staged to a tempfile then `tigris cp`)")
+@click.pass_context
+def put_object(ctx, bucket, key, file_path, text):
+    """Upload an object from a file or inline text."""
+    backend: TigrisBackend = ctx.obj["backend"]
+    use_json = ctx.obj.get("json", False)
+    skin = ctx.obj.get("skin")
+    if (file_path is None) == (text is None):
+        msg = "Provide exactly one of --file or --text"
+        _emit_error(use_json, skin, msg)
+        raise SystemExit(2)
+    try:
+        if file_path:
+            backend.put_object_from_file(bucket, key, file_path)
+        else:
+            backend.put_object_inline(bucket, key, text)
+        if use_json:
+            click.echo(json_mod.dumps(
+                {"bucket": bucket, "key": key, "status": "uploaded"}, indent=2
+            ))
+        else:
+            skin.success(f"Uploaded {bucket}/{key}")
+    except TigrisCliError as e:
+        _emit_error(use_json, skin, f"Failed to upload: {e}")
+        raise SystemExit(1)
+
+
+@object_group.command("get")
+@click.option("--bucket", required=True, help="Bucket name")
+@click.option("--key", required=True, help="Object key")
+@click.option("--output", required=True,
+              help="Local path to write the object to")
+@click.pass_context
+def get_object(ctx, bucket, key, output):
+    """Download an object to a local file (`tigris cp`)."""
+    backend: TigrisBackend = ctx.obj["backend"]
+    use_json = ctx.obj.get("json", False)
+    skin = ctx.obj.get("skin")
+    try:
+        backend.get_object_to_file(bucket, key, output)
+        if use_json:
+            click.echo(json_mod.dumps(
+                {"bucket": bucket, "key": key, "path": output}, indent=2
+            ))
+        else:
+            skin.success(f"Downloaded {bucket}/{key} -> {output}")
+    except TigrisCliError as e:
+        _emit_error(use_json, skin, f"Failed to download: {e}")
+        raise SystemExit(1)
+
+
+@object_group.command("delete")
+@click.option("--bucket", required=True, help="Bucket name")
+@click.option("--key", required=True, help="Object key")
+@click.pass_context
+def delete_object(ctx, bucket, key):
+    """Delete an object (`tigris rm`)."""
+    backend: TigrisBackend = ctx.obj["backend"]
+    use_json = ctx.obj.get("json", False)
+    skin = ctx.obj.get("skin")
+    try:
+        backend.delete_object(bucket, key)
+        if use_json:
+            click.echo(json_mod.dumps(
+                {"bucket": bucket, "key": key, "status": "deleted"}, indent=2
+            ))
+        else:
+            skin.success(f"Deleted {bucket}/{key}")
+    except TigrisCliError as e:
+        _emit_error(use_json, skin, f"Failed to delete: {e}")
+        raise SystemExit(1)
+
+
+@object_group.command("info")
+@click.option("--bucket", required=True, help="Bucket name")
+@click.option("--key", required=True, help="Object key")
+@click.pass_context
+def object_info(ctx, bucket, key):
+    """Get object metadata (`tigris stat`)."""
+    backend: TigrisBackend = ctx.obj["backend"]
+    use_json = ctx.obj.get("json", False)
+    skin = ctx.obj.get("skin")
+    try:
+        info = backend.head_object(bucket, key)
+        if use_json:
+            click.echo(json_mod.dumps(info or {"bucket": bucket, "key": key}, indent=2))
+        else:
+            skin.section(f"Object: {bucket}/{key}")
+            if isinstance(info, dict):
+                for k, v in info.items():
+                    skin.status(k, str(v))
+            else:
+                click.echo(info)
+    except TigrisCliError as e:
+        _emit_error(use_json, skin, f"Failed to get object info: {e}")
+        raise SystemExit(1)
+
+
+@object_group.command("cp")
+@click.argument("src")
+@click.argument("dst")
+@click.option("--recursive", "-r", is_flag=True, help="Copy directories recursively")
+@click.pass_context
+def copy_object(ctx, src, dst, recursive):
+    """Copy. SRC and DST are local paths or t3://bucket/key (or tigris://...).
+
+    At least one side must be a t3:// or tigris:// URI.
+    """
+    backend: TigrisBackend = ctx.obj["backend"]
+    use_json = ctx.obj.get("json", False)
+    skin = ctx.obj.get("skin")
+
+    if not (_is_remote(src) or _is_remote(dst)):
+        _emit_error(use_json, skin,
+                    "At least one of SRC or DST must be t3:// or tigris://")
+        raise SystemExit(2)
+
+    try:
+        backend.cp(src, dst, recursive=recursive)
+        if use_json:
+            click.echo(json_mod.dumps(
+                {"src": src, "dst": dst, "status": "copied"}, indent=2
+            ))
+        else:
+            skin.success(f"Copied {src} -> {dst}")
+    except TigrisCliError as e:
+        _emit_error(use_json, skin, f"Failed to copy: {e}")
+        raise SystemExit(1)
+
+
+def _emit_error(use_json: bool, skin, message: str) -> None:
+    if use_json:
+        click.echo(json_mod.dumps({"error": message}, indent=2))
+    elif skin:
+        skin.error(message)
+    else:
+        click.echo(message, err=True)

tigris/agent-harness/cli_anything/tigris/core/presign.py

@@ -0,0 +1,66 @@
+"""Presigned URL commands -- get, put.
+
+Wraps `tigris presign <path> --method get|put --expires-in <sec>`.
+"""
+
+import json as json_mod
+import click
+
+from ..utils.tigris_backend import TigrisBackend, TigrisCliError
+
+
+@click.group("presign")
+@click.pass_context
+def presign_group(ctx):
+    """Generate presigned URLs (wraps `tigris presign`)."""
+    pass
+
+
+def _presign(ctx, bucket: str, key: str, method: str, expires: int,
+             access_key: str | None) -> None:
+    backend: TigrisBackend = ctx.obj["backend"]
+    use_json = ctx.obj.get("json", False)
+    skin = ctx.obj.get("skin")
+    try:
+        url = backend.presign(bucket, key, method=method,
+                              expires_in=expires, access_key=access_key)
+        if use_json:
+            click.echo(json_mod.dumps(
+                {"url": url, "method": method.upper(), "expires_in": expires},
+                indent=2,
+            ))
+        else:
+            skin.success(f"Presigned {method.upper()} for {bucket}/{key} ({expires}s)")
+            click.echo(url)
+    except TigrisCliError as e:
+        if use_json:
+            click.echo(json_mod.dumps({"error": str(e)}, indent=2))
+        else:
+            skin.error(f"Failed to presign: {e}")
+        raise SystemExit(1)
+
+
+@presign_group.command("get")
+@click.option("--bucket", required=True, help="Bucket name")
+@click.option("--key", required=True, help="Object key")
+@click.option("--expires", default=3600, type=int,
+              help="URL lifetime in seconds (default: 3600)")
+@click.option("--access-key", default=None,
+              help="Access key ID to sign with (default: resolved automatically)")
+@click.pass_context
+def presign_get(ctx, bucket, key, expires, access_key):
+    """Presigned URL for downloading an object."""
+    _presign(ctx, bucket, key, "get", expires, access_key)
+
+
+@presign_group.command("put")
+@click.option("--bucket", required=True, help="Bucket name")
+@click.option("--key", required=True, help="Object key")
+@click.option("--expires", default=3600, type=int,
+              help="URL lifetime in seconds (default: 3600)")
+@click.option("--access-key", default=None,
+              help="Access key ID to sign with (default: resolved automatically)")
+@click.pass_context
+def presign_put(ctx, bucket, key, expires, access_key):
+    """Presigned URL for uploading an object."""
+    _presign(ctx, bucket, key, "put", expires, access_key)

tigris/agent-harness/cli_anything/tigris/core/snapshot.py

@@ -0,0 +1,82 @@
+"""Snapshot commands -- list, take.
+
+Wraps `tigris snapshots list/take`. Snapshots are point-in-time, read-only
+copies of a bucket's state — one of Tigris's agent-storage primitives that
+generic S3-compatible providers don't ship.
+"""
+
+import json as json_mod
+import click
+
+from ..utils.tigris_backend import TigrisBackend, TigrisCliError
+
+
+@click.group("snapshot")
+@click.pass_context
+def snapshot_group(ctx):
+    """Manage bucket snapshots (wraps `tigris snapshots`)."""
+    pass
+
+
+@snapshot_group.command("list")
+@click.argument("bucket")
+@click.pass_context
+def list_snapshots(ctx, bucket):
+    """List snapshots for a bucket."""
+    backend: TigrisBackend = ctx.obj["backend"]
+    use_json = ctx.obj.get("json", False)
+    skin = ctx.obj.get("skin")
+    try:
+        snaps = backend.list_snapshots(bucket)
+        if use_json:
+            click.echo(json_mod.dumps(snaps, indent=2))
+        else:
+            if not snaps:
+                skin.info(f"No snapshots found for '{bucket}'.")
+                return
+            if isinstance(snaps, list):
+                headers = ["Name / ID", "Created"]
+                rows = []
+                for s in snaps:
+                    if not isinstance(s, dict):
+                        rows.append([str(s), ""])
+                        continue
+                    rows.append([
+                        str(s.get("name") or s.get("id") or "?"),
+                        str(s.get("created") or s.get("createdAt") or ""),
+                    ])
+                skin.table(headers, rows)
+            else:
+                click.echo(snaps)
+    except TigrisCliError as e:
+        _emit_error(use_json, skin, f"Failed to list snapshots: {e}")
+        raise SystemExit(1)
+
+
+@snapshot_group.command("take")
+@click.argument("bucket")
+@click.option("--name", default=None, help="Optional snapshot label/name")
+@click.pass_context
+def take_snapshot(ctx, bucket, name):
+    """Take a point-in-time snapshot of a bucket."""
+    backend: TigrisBackend = ctx.obj["backend"]
+    use_json = ctx.obj.get("json", False)
+    skin = ctx.obj.get("skin")
+    try:
+        result = backend.take_snapshot(bucket, name=name)
+        if use_json:
+            click.echo(json_mod.dumps(result or {"bucket": bucket, "status": "snapshot_taken"}, indent=2))
+        else:
+            skin.success(f"Snapshot taken for '{bucket}'" + (f" ({name})" if name else ""))
+    except TigrisCliError as e:
+        _emit_error(use_json, skin, f"Failed to take snapshot: {e}")
+        raise SystemExit(1)
+
+
+def _emit_error(use_json: bool, skin, message: str) -> None:
+    if use_json:
+        click.echo(json_mod.dumps({"error": message}, indent=2))
+    elif skin:
+        skin.error(message)
+    else:
+        click.echo(message, err=True)

tigris/agent-harness/cli_anything/tigris/skills/SKILL.md

@@ -0,0 +1,199 @@
+---
+name: "cli-anything-tigris"
+description: >-
+  Command-line interface for Tigris object storage — wraps the official `tigris` CLI to expose buckets, objects, presigned URLs, snapshots, IAM, and scoped access keys to AI agents. Globally distributed, S3-compatible, no egress fees.
+---
+
+# cli-anything-tigris
+
+A stateless command-line interface for [Tigris](https://www.tigrisdata.com)
+object storage. Wraps the official `tigris` CLI so every Tigris primitive
+(snapshots, IAM, scoped credentials, OAuth) is reachable through a single
+agent-friendly entry point with `--json` everywhere.
+
+## Scope
+
+This harness is for the official Tigris CLI only. It shells out to the
+`tigris` binary and assumes Tigris auth through `tigris login` or Tigris
+access keys. It is not a generic S3 endpoint tool, and it does not manage
+MinIO, Cloudflare R2, AWS S3, or arbitrary S3-compatible endpoints.
+
+## Installation
+
+```bash
+# 1. Install the underlying Tigris CLI
+npm install -g @tigrisdata/cli
+# or:
+brew install tigrisdata/tap/tigris
+
+# 2. Authenticate (browser OAuth)
+tigris login
+
+# 3. Install this harness
+pip install cli-anything-tigris
+```
+
+**Prerequisites:**
+- Python 3.10+
+- `tigris` CLI on PATH (the binary's alias is `t3`)
+
+## Usage
+
+### Basic Commands
+
+```bash
+# Show help
+cli-anything-tigris --help
+
+# Start interactive REPL
+cli-anything-tigris
+
+# Whoami (JSON output for agents)
+cli-anything-tigris --json auth whoami
+
+# List buckets
+cli-anything-tigris --json bucket list
+
+# Upload a local file
+cli-anything-tigris --json object put --bucket my-bucket --key path/to/file.txt --file ./local.txt
+
+# Download an object
+cli-anything-tigris --json object get --bucket my-bucket --key path/to/file.txt --output ./out.txt
+
+# Server-side copy
+cli-anything-tigris --json object cp t3://my-bucket/src.txt t3://my-bucket/dst.txt
+
+# Take a snapshot
+cli-anything-tigris --json snapshot take my-bucket --name baseline-v1
+
+# Delete a bucket; --yes is required
+cli-anything-tigris --json bucket delete --name old-bucket --yes
+
+# Create a scoped access key for an agent run
+cli-anything-tigris --json access-key create my-agent-key
+cli-anything-tigris --json access-key assign tid_AaBb --bucket my-bucket --role Editor
+
+# Rotate or delete access keys; --yes is required
+cli-anything-tigris --json access-key rotate tid_AaBb --yes
+cli-anything-tigris --json access-key delete tid_AaBb --yes
+
+# Presigned download URL (1 hour)
+cli-anything-tigris --json presign get --bucket my-bucket --key path/to/file.txt --expires 3600
+```
+
+### REPL Mode
+
+When invoked without a subcommand, the CLI enters an interactive REPL with
+tab-completion and history.
+
+## Command Groups
+
+### auth
+OAuth-based authentication.
+
+| Command | Description |
+|---------|-------------|
+| `login` | Browser OAuth login (`tigris login`) |
+| `logout` | Log out of current session |
+| `whoami` | Print authenticated user / org |
+
+### bucket
+Bucket CRUD.
+
+| Command | Description |
+|---------|-------------|
+| `list` | List all buckets |
+| `create --name NAME` | Create a bucket |
+| `delete --name NAME --yes` | Delete an empty bucket (`--yes` required) |
+| `info NAME` | Get bucket info |
+
+### object
+Object operations (wraps `tigris ls/cp/rm/stat`).
+
+| Command | Description |
+|---------|-------------|
+| `list --bucket B [--prefix P] [--limit N]` | List objects |
+| `put --bucket B --key K (--file F \| --text T)` | Upload an object |
+| `get --bucket B --key K --output F` | Download an object |
+| `delete --bucket B --key K` | Delete an object |
+| `info --bucket B --key K` | Object metadata (HEAD / stat) |
+| `cp SRC DST [-r]` | Copy. Accepts `t3://` or `tigris://` URIs. |
+
+### presign
+Time-limited URLs.
+
+| Command | Description |
+|---------|-------------|
+| `get --bucket B --key K [--expires SEC]` | Presigned download URL |
+| `put --bucket B --key K [--expires SEC]` | Presigned upload URL |
+
+### snapshot
+Point-in-time bucket snapshots — a Tigris-specific primitive.
+
+| Command | Description |
+|---------|-------------|
+| `list BUCKET` | List snapshots for a bucket |
+| `take BUCKET [--name N]` | Take a snapshot |
+
+### access-key
+Scoped programmatic credentials — combine with `snapshot` for per-agent isolation.
+
+| Command | Description |
+|---------|-------------|
+| `list` | List all access keys |
+| `create NAME` | Create a new access key (secret shown ONCE) |
+| `get KEY_ID` | Show key details |
+| `delete KEY_ID --yes` | Permanently delete a key (`--yes` required) |
+| `assign KEY_ID --bucket B --role R` | Scope a key to a bucket + role |
+| `rotate KEY_ID --yes` | Rotate a key's secret (`--yes` required) |
+
+### iam
+Policies and organization users.
+
+| Command | Description |
+|---------|-------------|
+| `policy list` | List IAM policies |
+| `policy create NAME --document FILE` | Create a policy from a JSON file |
+| `user list` | List org users |
+| `user invite EMAIL [--role R]` | Invite a user |
+
+## Output Formats
+
+All commands support dual output modes:
+
+- **Human-readable** (default): tables, colors, formatted text
+- **Machine-readable** (`--json`): JSON envelope (or upstream CLI's
+  `--format json` output, echoed verbatim)
+
+## For AI Agents
+
+When using this CLI programmatically:
+
+1. Always pass `--json` for parseable output.
+2. Check return codes — 0 for success, non-zero for errors.
+3. Read stderr for error messages.
+4. `object cp` accepts `t3://bucket/key` or `tigris://bucket/key` URIs;
+   server-side copies (t3 → t3) skip the round trip entirely.
+5. `presign` returns a URL on stdout in human mode; in JSON mode it's the
+   `url` field.
+6. **For destructive work**: bucket deletion, access-key deletion, and
+   access-key rotation require explicit `--yes`; take a `snapshot` of the
+   target bucket first, then do the work, then either keep the snapshot or
+   discard.
+7. **For per-agent isolation**: `access-key create` + `access-key assign
+   --bucket B --role Editor` to mint a key scoped to one bucket; revoke with
+   `access-key delete` when the agent run ends.
+
+## Why Tigris
+
+- **Globally distributed.** Data placed close to wherever it's read.
+- **No egress fees.** Agents pulling artifacts from anywhere don't incur
+  per-region bandwidth charges.
+- **Snapshots + scoped credentials.** Primitives generic S3-compatible
+  providers don't ship — the foundation for per-agent isolation.
+- **S3-compatible.** Useful alongside S3-aware tools, but this harness
+  itself is not a generic S3/MinIO/R2/AWS endpoint manager.
+
+## Version
+
+1.0.0

tigris/agent-harness/cli_anything/tigris/tests/TEST.md

@@ -0,0 +1,65 @@
+# Tigris CLI Harness — Test Plan
+
+## Layout
+
+- `test_core.py` — unit tests for the `TigrisBackend` subprocess wrapper and
+  the Click CLI. `subprocess.run` is fully mocked, and `shutil.which` is
+  patched at module load so the backend believes `tigris` is on PATH. These
+  tests run with **no `tigris` CLI installed and no network access** — safe
+  for any CI environment.
+- `test_full_e2e.py` — real-world tests that shell out to the actual `tigris`
+  CLI against a real bucket. **Skipped by default**; gated on the
+  `CLI_ANYTHING_TIGRIS_RUN_E2E` env var plus `tigris` being on PATH plus a
+  configured test bucket.
+
+## Running unit tests
+
+```bash
+cd tigris/agent-harness
+pip install -e .[dev]
+pytest cli_anything/tigris/tests/test_core.py -v
+```
+
+All tests should pass with no Tigris CLI and no credentials.
+
+## Coverage areas (unit tests)
+
+| Area | Tests |
+|------|-------|
+| URI/path helpers (`_path_to_t3`, `_parse_tigris_uri`) | scheme normalization, happy + reject paths |
+| Backend init | binary resolution via `shutil.which`, missing-binary error, env-var export for credentials |
+| Bucket ops | `list/create/delete/head` invoke `tigris buckets …` with correct args + `--format json`; `delete` only passes `--yes` when explicitly requested |
+| Object ops | `list` (with prefix + client-side limit), `cp` (with `--recursive`), `put_from_file`, `put_inline` (tempfile path), `delete` (uses `rm --yes`), `head` (uses `stat`) |
+| Presign | flags forwarded, URL extracted from JSON dict, fallback parse from raw string |
+| Snapshots | `list/take` invoke `tigris snapshots …` |
+| Access keys | `list/create/get/delete/assign/rotate` flag wiring; `delete` and `rotate` only pass `--yes` when explicitly requested |
+| IAM | `policies list/create`, `users list/invite` flag wiring |
+| Error path | non-zero exit code raises `TigrisCliError` with stderr |
+| CLI integration | `--json` output works for bucket/object/presign/snapshot/access-key/auth; `object put` without `--file`/`--text` errors; `object cp` with no `t3://` errors |
+
+## Running e2e tests
+
+```bash
+# Install + auth
+npm install -g @tigrisdata/cli     # or: brew install tigrisdata/tap/tigris
+tigris login
+
+# Configure
+export CLI_ANYTHING_TIGRIS_TEST_BUCKET=<your-test-bucket>
+export CLI_ANYTHING_TIGRIS_RUN_E2E=1
+
+# Run
+pytest cli_anything/tigris/tests/test_full_e2e.py -v
+```
+
+E2e tests cover:
+
+- `whoami` returns a non-empty session
+- `list_buckets` includes the configured test bucket
+- put / get / head / list / delete round trip for a single object
+- presigned URL is well-formed and includes the key's last segment
+- `snapshots list` succeeds against the test bucket
+
+Each test uses a per-run UUID prefix for keys to avoid collisions when
+multiple devs / CI runners hit the same bucket. Cleanup happens in
+`finally` blocks so partial failures don't leave litter.

tigris/agent-harness/cli_anything/tigris/tests/test_core.py

@@ -0,0 +1,441 @@
+"""Unit tests for the Tigris CLI harness.
+
+`subprocess.run` is fully mocked so these tests run without the `tigris` CLI
+installed or any network access. End-to-end tests against a real Tigris CLI
+live in `test_full_e2e.py`.
+"""
+
+import json as json_mod
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+import pytest
+from click.testing import CliRunner
+
+# Patch shutil.which BEFORE the backend is constructed so the resolver
+# always claims `tigris` is available.
+_WHICH_PATCH = patch(
+    "cli_anything.tigris.utils.tigris_backend.shutil.which",
+    return_value="/usr/local/bin/tigris",
+)
+_WHICH_PATCH.start()
+
+from cli_anything.tigris.tigris_cli import cli  # noqa: E402
+from cli_anything.tigris.utils.tigris_backend import (   # noqa: E402
+    TigrisBackend,
+    TigrisCliError,
+    _path_to_t3,
+)
+from cli_anything.tigris.core.object import _parse_tigris_uri  # noqa: E402
+
+
+# ── Helpers ───────────────────────────────────────────────────────────
+
+
+def _mock_run(stdout: str = "", stderr: str = "", returncode: int = 0):
+    """Build a mock subprocess.run that returns the given completed-process."""
+    return SimpleNamespace(stdout=stdout, stderr=stderr, returncode=returncode)
+
+
+def _patch_run(json_out=None, text_out="", returncode=0):
+    """Convenience: patch subprocess.run inside the backend module."""
+    if json_out is not None:
+        stdout = json_mod.dumps(json_out)
+    else:
+        stdout = text_out
+    return patch(
+        "cli_anything.tigris.utils.tigris_backend.subprocess.run",
+        return_value=_mock_run(stdout=stdout, returncode=returncode),
+    )
+
+
+# ── URI / path helpers ────────────────────────────────────────────────
+
+
+def test_path_to_t3_normalizes_schemes():
+    assert _path_to_t3("t3://b/k") == "t3://b/k"
+    assert _path_to_t3("tigris://b/k") == "t3://b/k"
+    assert _path_to_t3("b/k") == "t3://b/k"
+    assert _path_to_t3("/b/k") == "t3://b/k"
+
+
+def test_parse_tigris_uri_happy():
+    assert _parse_tigris_uri("t3://b/k") == ("b", "k")
+    assert _parse_tigris_uri("tigris://b/a/b/c") == ("b", "a/b/c")
+
+
+def test_parse_tigris_uri_rejects_bad():
+    import click as _click
+    for bad in ("s3://b/k", "tigris://nokey", "t3:///nobucket", "plainpath"):
+        with pytest.raises(_click.UsageError):
+            _parse_tigris_uri(bad)
+
+
+# ── Backend init ──────────────────────────────────────────────────────
+
+
+def test_backend_init_resolves_tigris_path():
+    # _WHICH_PATCH at module scope makes this succeed.
+    b = TigrisBackend()
+    assert b.cli_path == "/usr/local/bin/tigris"
+
+
+def test_backend_init_raises_when_tigris_missing():
+    with patch(
+        "cli_anything.tigris.utils.tigris_backend.shutil.which",
+        return_value=None,
+    ):
+        with pytest.raises(TigrisCliError, match="not found on PATH"):
+            TigrisBackend()
+
+
+def test_backend_extra_env_set_when_credentials_passed():
+    b = TigrisBackend(access_key="ak", secret_key="sk")
+    assert b._extra_env["TIGRIS_STORAGE_ACCESS_KEY_ID"] == "ak"
+    assert b._extra_env["AWS_ACCESS_KEY_ID"] == "ak"
+    assert b._extra_env["TIGRIS_STORAGE_SECRET_ACCESS_KEY"] == "sk"
+    assert b._extra_env["AWS_SECRET_ACCESS_KEY"] == "sk"
+
+
+# ── Backend method invocations (verify the args passed to subprocess) ──
+
+
+def _run_call_args(mock_run) -> list[str]:
+    """Extract the argv list passed to subprocess.run."""
+    args, _ = mock_run.call_args
+    return args[0]
+
+
+def test_list_buckets_invokes_correct_args():
+    with _patch_run(json_out=[{"name": "alpha"}]) as m:
+        b = TigrisBackend()
+        result = b.list_buckets()
+        assert result == [{"name": "alpha"}]
+        argv = _run_call_args(m)
+        assert argv[1:] == ["buckets", "list", "--format", "json"]
+
+
+def test_create_bucket_invokes_correct_args():
+    with _patch_run(json_out={"name": "x", "status": "created"}) as m:
+        b = TigrisBackend()
+        b.create_bucket("x")
+        assert _run_call_args(m)[1:] == ["buckets", "create", "x", "--format", "json"]
+
+
+def test_delete_bucket_defaults_to_confirmation():
+    with _patch_run(json_out={"status": "deleted"}) as m:
+        b = TigrisBackend()
+        b.delete_bucket("x")
+        argv = _run_call_args(m)
+        assert "--yes" not in argv
+        assert "buckets" in argv and "delete" in argv and "x" in argv
+
+
+def test_delete_bucket_includes_yes_flag_when_requested():
+    with _patch_run(json_out={"status": "deleted"}) as m:
+        b = TigrisBackend()
+        b.delete_bucket("x", yes=True)
+        argv = _run_call_args(m)
+        assert "--yes" in argv
+        assert "buckets" in argv and "delete" in argv and "x" in argv
+
+
+def test_head_bucket_uses_get():
+    with _patch_run(json_out={"name": "x"}) as m:
+        b = TigrisBackend()
+        b.head_bucket("x")
+        assert _run_call_args(m)[1:] == ["buckets", "get", "x", "--format", "json"]
+
+
+def test_list_objects_with_prefix_and_limit():
+    payload = [{"key": "foo"}, {"key": "bar"}, {"key": "baz"}]
+    with _patch_run(json_out=payload) as m:
+        b = TigrisBackend()
+        result = b.list_objects("my-bucket", prefix="dir/", limit=2)
+        argv = _run_call_args(m)
+        assert argv[1:] == ["ls", "t3://my-bucket/dir", "--format", "json"]
+        assert result == payload[:2]  # client-side limit applied
+
+
+def test_cp_uses_recursive_flag_when_set():
+    with _patch_run(text_out="ok") as m:
+        b = TigrisBackend()
+        b.cp("./local/", "t3://my-bucket/dst/", recursive=True)
+        argv = _run_call_args(m)
+        assert "--recursive" in argv
+        # JSON-mode is OFF for cp (no --format json appended)
+        assert "--format" not in argv
+
+
+def test_put_object_from_file_uses_cp():
+    with _patch_run(text_out="") as m:
+        b = TigrisBackend()
+        b.put_object_from_file("my-bucket", "k", "./local.txt")
+        argv = _run_call_args(m)
+        assert argv[1] == "cp"
+        assert argv[2] == "./local.txt"
+        assert argv[3] == "t3://my-bucket/k"
+
+
+def test_put_object_inline_writes_tempfile_then_cleans_up():
+    with _patch_run(text_out="") as m:
+        b = TigrisBackend()
+        b.put_object_inline("my-bucket", "k", "hello body")
+        argv = _run_call_args(m)
+        assert argv[1] == "cp"
+        # tmp path -> t3 url
+        assert argv[3] == "t3://my-bucket/k"
+
+
+def test_delete_object_uses_rm_with_yes():
+    with _patch_run(text_out="") as m:
+        b = TigrisBackend()
+        b.delete_object("my-bucket", "k")
+        argv = _run_call_args(m)
+        assert argv[1] == "rm"
+        assert argv[2] == "t3://my-bucket/k"
+        assert "--yes" in argv
+
+
+def test_head_object_uses_stat():
+    with _patch_run(json_out={"size": 42}) as m:
+        b = TigrisBackend()
+        info = b.head_object("my-bucket", "k")
+        argv = _run_call_args(m)
+        assert argv[1:] == ["stat", "t3://my-bucket/k", "--format", "json"]
+        assert info == {"size": 42}
+
+
+def test_presign_returns_url_from_dict():
+    with _patch_run(json_out={"url": "https://signed/url"}) as m:
+        b = TigrisBackend()
+        url = b.presign("my-bucket", "k", method="get", expires_in=600)
+        argv = _run_call_args(m)
+        assert "presign" in argv
+        assert "t3://my-bucket/k" in argv
+        assert "--method" in argv and "get" in argv
+        assert "--expires-in" in argv and "600" in argv
+        assert url == "https://signed/url"
+
+
+def test_presign_falls_back_to_string():
+    with _patch_run(text_out="https://signed/url\n") as m:
+        b = TigrisBackend()
+        # text_out is returned through JSONDecodeError fallback
+        url = b.presign("my-bucket", "k")
+        assert url == "https://signed/url"
+
+
+# ── Snapshots ─────────────────────────────────────────────────────────
+
+
+def test_list_snapshots_invokes_correct_args():
+    with _patch_run(json_out=[]) as m:
+        b = TigrisBackend()
+        b.list_snapshots("my-bucket")
+        assert _run_call_args(m)[1:] == ["snapshots", "list", "my-bucket", "--format", "json"]
+
+
+def test_take_snapshot_with_name():
+    with _patch_run(json_out={"status": "ok"}) as m:
+        b = TigrisBackend()
+        b.take_snapshot("my-bucket", name="v1")
+        argv = _run_call_args(m)
+        assert "snapshots" in argv and "take" in argv
+        assert "--name" in argv and "v1" in argv
+
+
+# ── Access keys ───────────────────────────────────────────────────────
+
+
+def test_list_access_keys():
+    with _patch_run(json_out=[]) as m:
+        b = TigrisBackend()
+        b.list_access_keys()
+        assert _run_call_args(m)[1:] == ["access-keys", "list", "--format", "json"]
+
+
+def test_delete_access_key_defaults_to_confirmation():
+    with _patch_run(json_out={"status": "deleted"}) as m:
+        b = TigrisBackend()
+        b.delete_access_key("tid_AaBb")
+        argv = _run_call_args(m)
+        assert "--yes" not in argv
+        assert "access-keys" in argv and "delete" in argv and "tid_AaBb" in argv
+
+
+def test_delete_access_key_includes_yes_flag_when_requested():
+    with _patch_run(json_out={"status": "deleted"}) as m:
+        b = TigrisBackend()
+        b.delete_access_key("tid_AaBb", yes=True)
+        argv = _run_call_args(m)
+        assert "--yes" in argv
+        assert "access-keys" in argv and "delete" in argv and "tid_AaBb" in argv
+
+
+def test_assign_access_key_uses_role_and_bucket():
+    with _patch_run(json_out={"status": "assigned"}) as m:
+        b = TigrisBackend()
+        b.assign_access_key("tid_AaBb", bucket="my-bucket", role="Editor")
+        argv = _run_call_args(m)
+        assert "access-keys" in argv and "assign" in argv and "tid_AaBb" in argv
+        assert "--bucket" in argv and "my-bucket" in argv
+        assert "--role" in argv and "Editor" in argv
+
+
+def test_rotate_access_key_defaults_to_confirmation():
+    with _patch_run(json_out={"status": "rotated"}) as m:
+        b = TigrisBackend()
+        b.rotate_access_key("tid_AaBb")
+        argv = _run_call_args(m)
+        assert "--yes" not in argv
+        assert "access-keys" in argv and "rotate" in argv and "tid_AaBb" in argv
+
+
+def test_rotate_access_key_includes_yes_flag_when_requested():
+    with _patch_run(json_out={"status": "rotated"}) as m:
+        b = TigrisBackend()
+        b.rotate_access_key("tid_AaBb", yes=True)
+        argv = _run_call_args(m)
+        assert "--yes" in argv
+        assert "access-keys" in argv and "rotate" in argv and "tid_AaBb" in argv
+
+
+# ── IAM ───────────────────────────────────────────────────────────────
+
+
+def test_list_iam_policies():
+    with _patch_run(json_out=[]) as m:
+        b = TigrisBackend()
+        b.list_iam_policies()
+        assert _run_call_args(m)[1:] == ["iam", "policies", "list", "--format", "json"]
+
+
+def test_invite_iam_user():
+    with _patch_run(json_out={"status": "invited"}) as m:
+        b = TigrisBackend()
+        b.invite_iam_user("user@example.com", role="admin")
+        argv = _run_call_args(m)
+        assert "iam" in argv and "users" in argv and "invite" in argv
+        assert "user@example.com" in argv
+        assert "--role" in argv and "admin" in argv
+
+
+# ── Error handling ────────────────────────────────────────────────────
+
+
+def test_nonzero_exit_raises_tigris_cli_error():
+    with patch(
+        "cli_anything.tigris.utils.tigris_backend.subprocess.run",
+        return_value=_mock_run(stderr="boom", returncode=1),
+    ):
+        b = TigrisBackend()
+        with pytest.raises(TigrisCliError, match="boom"):
+            b.list_buckets()
+
+
+# ── CLI integration tests (backend uses mocked subprocess) ────────────
+
+
+def test_cli_bucket_list_json():
+    with _patch_run(json_out=[{"name": "demo"}]):
+        runner = CliRunner()
+        r = runner.invoke(cli, ["--json", "bucket", "list"])
+        assert r.exit_code == 0, r.output
+        assert "demo" in r.output
+
+
+def test_cli_bucket_delete_requires_yes():
+    with patch("cli_anything.tigris.utils.tigris_backend.subprocess.run") as m:
+        runner = CliRunner()
+        r = runner.invoke(cli, ["--json", "bucket", "delete", "--name", "b"])
+        assert r.exit_code != 0
+        assert "without --yes" in r.output
+        m.assert_not_called()
+
+
+def test_cli_bucket_delete_with_yes_passes_yes_flag():
+    with _patch_run(json_out={"status": "deleted"}) as m:
+        runner = CliRunner()
+        r = runner.invoke(cli, ["--json", "bucket", "delete", "--name", "b", "--yes"])
+        assert r.exit_code == 0, r.output
+        assert "--yes" in _run_call_args(m)
+
+
+def test_cli_object_put_requires_file_or_text():
+    runner = CliRunner()
+    r = runner.invoke(cli, ["--json", "object", "put", "--bucket", "b", "--key", "k"])
+    assert r.exit_code != 0
+
+
+def test_cli_object_cp_requires_one_remote():
+    runner = CliRunner()
+    r = runner.invoke(cli, ["--json", "object", "cp", "./a", "./b"])
+    assert r.exit_code != 0
+
+
+def test_cli_presign_get_json():
+    with _patch_run(json_out={"url": "https://signed"}):
+        runner = CliRunner()
+        r = runner.invoke(cli, ["--json", "presign", "get", "--bucket", "b", "--key", "k"])
+        assert r.exit_code == 0, r.output
+        assert "https://signed" in r.output
+
+
+def test_cli_snapshot_take_json():
+    with _patch_run(json_out={"status": "ok"}):
+        runner = CliRunner()
+        r = runner.invoke(cli, ["--json", "snapshot", "take", "my-bucket", "--name", "v1"])
+        assert r.exit_code == 0, r.output
+
+
+def test_cli_access_key_assign_json():
+    with _patch_run(json_out={"status": "assigned"}):
+        runner = CliRunner()
+        r = runner.invoke(cli, [
+            "--json", "access-key", "assign", "tid_x",
+            "--bucket", "b", "--role", "Editor",
+        ])
+        assert r.exit_code == 0, r.output
+
+
+def test_cli_access_key_delete_requires_yes():
+    with patch("cli_anything.tigris.utils.tigris_backend.subprocess.run") as m:
+        runner = CliRunner()
+        r = runner.invoke(cli, ["--json", "access-key", "delete", "tid_x"])
+        assert r.exit_code != 0
+        assert "without --yes" in r.output
+        m.assert_not_called()
+
+
+def test_cli_access_key_delete_with_yes_passes_yes_flag():
+    with _patch_run(json_out={"status": "deleted"}) as m:
+        runner = CliRunner()
+        r = runner.invoke(cli, ["--json", "access-key", "delete", "tid_x", "--yes"])
+        assert r.exit_code == 0, r.output
+        assert "--yes" in _run_call_args(m)
+
+
+def test_cli_access_key_rotate_requires_yes():
+    with patch("cli_anything.tigris.utils.tigris_backend.subprocess.run") as m:
+        runner = CliRunner()
+        r = runner.invoke(cli, ["--json", "access-key", "rotate", "tid_x"])
+        assert r.exit_code != 0
+        assert "without --yes" in r.output
+        m.assert_not_called()
+
+
+def test_cli_access_key_rotate_with_yes_passes_yes_flag():
+    with _patch_run(json_out={"status": "rotated"}) as m:
+        runner = CliRunner()
+        r = runner.invoke(cli, ["--json", "access-key", "rotate", "tid_x", "--yes"])
+        assert r.exit_code == 0, r.output
+        assert "--yes" in _run_call_args(m)
+
+
+def test_cli_auth_whoami_json():
+    with _patch_run(json_out={"user": "dave", "org": "tigris"}):
+        runner = CliRunner()
+        r = runner.invoke(cli, ["--json", "auth", "whoami"])
+        assert r.exit_code == 0, r.output
+        assert "dave" in r.output

tigris/agent-harness/cli_anything/tigris/tests/test_full_e2e.py

@@ -0,0 +1,110 @@
+"""End-to-end tests against a real Tigris CLI and a real bucket.
+
+These tests are SKIPPED by default. To run them locally:
+
+    npm install -g @tigrisdata/cli     # or: brew install tigrisdata/tap/tigris
+    tigris login                       # one-time OAuth setup
+
+    export CLI_ANYTHING_TIGRIS_TEST_BUCKET=<a bucket you can write to>
+    export CLI_ANYTHING_TIGRIS_RUN_E2E=1
+
+    pytest cli_anything/tigris/tests/test_full_e2e.py -v
+
+Test objects live under a per-run UUID prefix and are cleaned up in
+teardown so concurrent runs do not collide.
+"""
+
+import os
+import shutil
+import uuid
+
+import pytest
+
+from cli_anything.tigris.utils.tigris_backend import TigrisBackend
+
+RUN_E2E = os.environ.get("CLI_ANYTHING_TIGRIS_RUN_E2E") == "1"
+TEST_BUCKET = os.environ.get("CLI_ANYTHING_TIGRIS_TEST_BUCKET")
+HAS_TIGRIS_CLI = shutil.which("tigris") is not None
+
+pytestmark = pytest.mark.skipif(
+    not (RUN_E2E and TEST_BUCKET and HAS_TIGRIS_CLI),
+    reason=(
+        "Set CLI_ANYTHING_TIGRIS_RUN_E2E=1 + "
+        "CLI_ANYTHING_TIGRIS_TEST_BUCKET=<bucket>, install the `tigris` CLI "
+        "(npm install -g @tigrisdata/cli), and run `tigris login`."
+    ),
+)
+
+
+@pytest.fixture(scope="module")
+def backend():
+    return TigrisBackend()
+
+
+@pytest.fixture
+def test_key():
+    return f"cli-anything-e2e/{uuid.uuid4()}.txt"
+
+
+def test_whoami_returns_user(backend):
+    info = backend.whoami()
+    # whoami output shape varies — just check it returned something truthy
+    assert info, f"whoami returned empty: {info!r}"
+
+
+def test_list_buckets_includes_test_bucket(backend):
+    buckets = backend.list_buckets()
+    if isinstance(buckets, list):
+        names = {
+            b.get("name") or b.get("Name")
+            for b in buckets if isinstance(b, dict)
+        }
+        assert TEST_BUCKET in names, f"{TEST_BUCKET} not in {names}"
+
+
+def test_put_get_delete_round_trip(backend, test_key, tmp_path):
+    body = "hello from cli-anything-tigris e2e\n"
+
+    # PUT via inline-text path
+    backend.put_object_inline(TEST_BUCKET, test_key, body)
+
+    try:
+        # GET to a local file and verify contents
+        out = tmp_path / "downloaded.txt"
+        backend.get_object_to_file(TEST_BUCKET, test_key, str(out))
+        assert out.read_text() == body
+
+        # HEAD / stat — returns *some* metadata
+        info = backend.head_object(TEST_BUCKET, test_key)
+        assert info, f"head_object returned empty: {info!r}"
+
+        # LIST narrowed by prefix should include our key
+        prefix = test_key.rsplit("/", 1)[0]
+        listing = backend.list_objects(TEST_BUCKET, prefix=prefix, limit=50)
+        if isinstance(listing, list):
+            found = any(
+                (o.get("key") or o.get("Key") or "") == test_key
+                for o in listing if isinstance(o, dict)
+            )
+            assert found, f"{test_key} not in {listing}"
+    finally:
+        backend.delete_object(TEST_BUCKET, test_key)
+
+
+def test_presigned_url_includes_key(backend, test_key):
+    backend.put_object_inline(TEST_BUCKET, test_key, "presign-test")
+    try:
+        url = backend.presign(TEST_BUCKET, test_key, method="get", expires_in=60)
+        assert isinstance(url, str) and url.startswith("http"), url
+        # Last path segment should appear in the URL
+        leaf = test_key.split("/")[-1]
+        assert leaf in url, f"{leaf} not in {url}"
+    finally:
+        backend.delete_object(TEST_BUCKET, test_key)
+
+
+def test_snapshot_list_works(backend):
+    snaps = backend.list_snapshots(TEST_BUCKET)
+    # We don't assert on count — just that the call succeeded and returned
+    # something parseable (list, dict, or string).
+    assert snaps is not None or snaps == []

tigris/agent-harness/cli_anything/tigris/tigris_cli.py

@@ -0,0 +1,157 @@
+"""Tigris CLI-Anything harness — Click CLI + REPL.
+
+Wraps the official Tigris CLI (`tigris`) so all of its primitives —
+including snapshots, IAM, scoped access keys, and OAuth — are reachable
+through a single agent-friendly entry point with `--json` everywhere.
+
+Tigris is a globally distributed, S3-compatible object storage service
+with no egress fees. https://www.tigrisdata.com
+"""
+
+import shlex
+
+import click
+
+from .utils.tigris_backend import TigrisBackend, TigrisCliError
+from .utils.repl_skin import ReplSkin
+from .core.auth import auth_group
+from .core.bucket import bucket_group
+from .core.object import object_group
+from .core.presign import presign_group
+from .core.snapshot import snapshot_group
+from .core.access_key import access_key_group
+from .core.iam import iam_group
+
+
+@click.group(invoke_without_command=True)
+@click.option("--json", "use_json", is_flag=True, default=False,
+              help="Output in JSON format")
+@click.option("--cli-path", default="tigris",
+              help="Path to the tigris CLI binary (default: 'tigris' on PATH)")
+@click.option("--access-key", default=None,
+              help="Optional access key ID (exported to env for child processes)")
+@click.option("--secret-key", default=None,
+              help="Optional secret key (exported to env for child processes)")
+@click.pass_context
+def cli(ctx, use_json, cli_path, access_key, secret_key):
+    """CLI-Anything harness for Tigris (S3-compatible object storage).
+
+    Wraps the official `tigris` CLI. Install it once with
+    `npm install -g @tigrisdata/cli` or `brew install tigrisdata/tap/tigris`,
+    run `tigris login`, then drive everything from here.
+    """
+    ctx.ensure_object(dict)
+    ctx.obj["json"] = use_json
+    try:
+        ctx.obj["backend"] = TigrisBackend(
+            cli_path=cli_path,
+            access_key=access_key,
+            secret_key=secret_key,
+        )
+    except TigrisCliError as e:
+        # Defer the error so `--help` still works without the binary.
+        ctx.obj["backend"] = None
+        ctx.obj["backend_error"] = str(e)
+    ctx.obj["skin"] = ReplSkin("tigris", version="1.0.0")
+
+    if ctx.invoked_subcommand is None:
+        _run_repl(ctx)
+
+
+cli.add_command(auth_group)
+cli.add_command(bucket_group)
+cli.add_command(object_group)
+cli.add_command(presign_group)
+cli.add_command(snapshot_group)
+cli.add_command(access_key_group)
+cli.add_command(iam_group)
+
+
+# ── REPL Commands Map (for help display) ─────────────────────────────
+
+_REPL_COMMANDS = {
+    "auth login":                                       "Browser OAuth login (`tigris login`)",
+    "auth logout":                                      "Log out of current session",
+    "auth whoami":                                      "Print current user / org",
+    "bucket list":                                      "List buckets",
+    "bucket create --name NAME":                        "Create a bucket",
+    "bucket delete --name NAME --yes":                  "Delete an empty bucket",
+    "bucket info NAME":                                 "Get bucket info",
+    "object list --bucket B [--prefix P]":              "List objects",
+    "object put --bucket B --key K --file F":           "Upload a file",
+    "object put --bucket B --key K --text T":           "Upload inline text",
+    "object get --bucket B --key K --output F":         "Download to file",
+    "object delete --bucket B --key K":                 "Delete an object",
+    "object info --bucket B --key K":                   "Object metadata (stat)",
+    "object cp SRC DST [-r]":                           "Copy local↔t3 or t3↔t3",
+    "presign get --bucket B --key K [--expires SEC]":   "Presigned GET URL",
+    "presign put --bucket B --key K [--expires SEC]":   "Presigned PUT URL",
+    "snapshot list BUCKET":                             "List bucket snapshots",
+    "snapshot take BUCKET [--name N]":                  "Take a point-in-time snapshot",
+    "access-key list":                                  "List access keys",
+    "access-key create NAME":                           "Create an access key",
+    "access-key get KEY_ID":                            "Get access key details",
+    "access-key delete KEY_ID --yes":                   "Delete an access key",
+    "access-key assign KEY_ID --bucket B --role R":     "Scope a key to a bucket/role",
+    "access-key rotate KEY_ID --yes":                   "Rotate a key's secret",
+    "iam policy list / create NAME --document F":       "Manage IAM policies",
+    "iam user list / invite EMAIL [--role R]":          "Manage org users",
+    "help":                                             "Show this help",
+    "quit / exit":                                      "Exit the REPL",
+}
+
+
+def _run_repl(ctx):
+    """Launch the interactive REPL."""
+    skin: ReplSkin = ctx.obj["skin"]
+    skin.print_banner()
+
+    if ctx.obj.get("backend") is None:
+        skin.error(ctx.obj.get("backend_error", "tigris CLI not available"))
+        return
+
+    session = skin.create_prompt_session()
+
+    while True:
+        try:
+            user_input = skin.get_input(session, context="tigris")
+        except (EOFError, KeyboardInterrupt):
+            skin.print_goodbye()
+            break
+
+        if not user_input:
+            continue
+
+        cmd = user_input.strip().lower()
+
+        if cmd in ("quit", "exit", "q"):
+            skin.print_goodbye()
+            break
+
+        if cmd in ("help", "h", "?"):
+            skin.help(_REPL_COMMANDS)
+            continue
+
+        try:
+            args = shlex.split(user_input)
+        except ValueError as e:
+            skin.error(f"Parse error: {e}")
+            continue
+
+        try:
+            cli.main(args=args, obj=ctx.obj, standalone_mode=False)
+        except SystemExit:
+            pass
+        except click.exceptions.UsageError as e:
+            skin.error(str(e))
+        except Exception as e:
+            skin.error(f"Error: {e}")
+
+
+def main():
+    """Entry point."""
+    cli(auto_envvar_prefix="TIGRIS_CLI")
+
+
+if __name__ == "__main__":
+    main()

tigris/agent-harness/cli_anything/tigris/utils/repl_skin.py

@@ -0,0 +1,567 @@
+"""cli-anything REPL Skin — Unified terminal interface for all CLI harnesses.
+
+Copy this file into your CLI package at:
+    cli_anything/<software>/utils/repl_skin.py
+
+Usage:
+    from cli_anything.<software>.utils.repl_skin import ReplSkin
+
+    skin = ReplSkin("shotcut", version="1.0.0")
+    skin.print_banner()  # auto-detects repo-root or packaged SKILL.md
+    prompt_text = skin.prompt(project_name="my_video.mlt", modified=True)
+    skin.success("Project saved")
+    skin.error("File not found")
+    skin.warning("Unsaved changes")
+    skin.info("Processing 24 clips...")
+    skin.status("Track 1", "3 clips, 00:02:30")
+    skin.table(headers, rows)
+    skin.print_goodbye()
+"""
+
+import os
+import sys
+from pathlib import Path
+
+# ── ANSI color codes (no external deps for core styling) ──────────────
+
+_RESET = "\033[0m"
+_BOLD = "\033[1m"
+_DIM = "\033[2m"
+_ITALIC = "\033[3m"
+_UNDERLINE = "\033[4m"
+
+# Brand colors
+_CYAN = "\033[38;5;80m"       # cli-anything brand cyan
+_CYAN_BG = "\033[48;5;80m"
+_WHITE = "\033[97m"
+_GRAY = "\033[38;5;245m"
+_DARK_GRAY = "\033[38;5;240m"
+_LIGHT_GRAY = "\033[38;5;250m"
+
+# Software accent colors — each software gets a unique accent
+_ACCENT_COLORS = {
+    "gimp":        "\033[38;5;214m",   # warm orange
+    "blender":     "\033[38;5;208m",   # deep orange
+    "inkscape":    "\033[38;5;39m",    # bright blue
+    "audacity":    "\033[38;5;33m",    # navy blue
+    "libreoffice": "\033[38;5;40m",    # green
+    "obs_studio":  "\033[38;5;55m",    # purple
+    "kdenlive":    "\033[38;5;69m",    # slate blue
+    "shotcut":     "\033[38;5;35m",    # teal green
+}
+_DEFAULT_ACCENT = "\033[38;5;75m"      # default sky blue
+
+# Status colors
+_GREEN = "\033[38;5;78m"
+_YELLOW = "\033[38;5;220m"
+_RED = "\033[38;5;196m"
+_BLUE = "\033[38;5;75m"
+_MAGENTA = "\033[38;5;176m"
+
+_SKILL_SOURCE_REPO = os.environ.get("CLI_ANYTHING_SKILL_REPO", "HKUDS/CLI-Anything")
+
+# ── Brand icon ────────────────────────────────────────────────────────
+
+# The cli-anything icon: a small colored diamond/chevron mark
+_ICON = f"{_CYAN}{_BOLD}◆{_RESET}"
+_ICON_SMALL = f"{_CYAN}▸{_RESET}"
+
+# ── Box drawing characters ────────────────────────────────────────────
+
+_H_LINE = "─"
+_V_LINE = "│"
+_TL = "╭"
+_TR = "╮"
+_BL = "╰"
+_BR = "╯"
+_T_DOWN = "┬"
+_T_UP = "┴"
+_T_RIGHT = "├"
+_T_LEFT = "┤"
+_CROSS = "┼"
+
+
+def _strip_ansi(text: str) -> str:
+    """Remove ANSI escape codes for length calculation."""
+    import re
+    return re.sub(r"\033\[[^m]*m", "", text)
+
+
+def _visible_len(text: str) -> int:
+    """Get visible length of text (excluding ANSI codes)."""
+    return len(_strip_ansi(text))
+
+
+def _display_home_path(path: str) -> str:
+    """Display a path relative to the home directory when possible."""
+    expanded = Path(path).expanduser().resolve()
+    home = Path.home().resolve()
+    try:
+        relative = expanded.relative_to(home)
+        return f"~/{relative.as_posix()}"
+    except ValueError:
+        return str(expanded)
+
+
+class ReplSkin:
+    """Unified REPL skin for cli-anything CLIs.
+
+    Provides consistent branding, prompts, and message formatting
+    across all CLI harnesses built with the cli-anything methodology.
+    """
+
+    def __init__(self, software: str, version: str = "1.0.0",
+                 history_file: str | None = None, skill_path: str | None = None):
+        """Initialize the REPL skin.
+
+        Args:
+            software: Software name (e.g., "gimp", "shotcut", "blender").
+            version: CLI version string.
+            history_file: Path for persistent command history.
+                         Defaults to ~/.cli-anything-<software>/history
+            skill_path: Path to the SKILL.md file for agent discovery.
+                        Auto-detected from the repo-root skills/ tree when present,
+                        otherwise from the package's skills/ directory.
+                        Displayed in banner for AI agents to know where to read skill info.
+        """
+        self.software = software.lower().replace("-", "_")
+        self.display_name = software.replace("_", " ").title()
+        self.version = version
+        software_aliases = {"iterm2_ctl": "iterm2"}
+        self.skill_slug = software_aliases.get(self.software, self.software).replace("_", "-")
+        self.skill_id = f"cli-anything-{self.skill_slug}"
+        self.skill_install_cmd = (
+            f"npx skills add {_SKILL_SOURCE_REPO} --skill {self.skill_id} -g -y"
+        )
+        global_skill_root = Path(
+            os.environ.get("CLI_ANYTHING_GLOBAL_SKILLS_DIR", str(Path.home() / ".agents" / "skills"))
+        ).expanduser()
+        self.global_skill_path = str(global_skill_root / self.skill_id / "SKILL.md")
+
+        # Prefer repo-root canonical skills/<skill-id>/SKILL.md when running
+        # inside the CLI-Anything monorepo. Fall back to the packaged
+        # cli_anything/<software>/skills/SKILL.md for installed harnesses.
+        if skill_path is None:
+            package_skill = Path(__file__).resolve().parent.parent / "skills" / "SKILL.md"
+            repo_skill = None
+            for parent in Path(__file__).resolve().parents:
+                candidate = parent / "skills" / self.skill_id / "SKILL.md"
+                if candidate.is_file():
+                    repo_skill = candidate
+                    break
+            if repo_skill and repo_skill.is_file():
+                skill_path = str(repo_skill)
+            elif package_skill.is_file():
+                skill_path = str(package_skill)
+        self.skill_path = skill_path
+        self.accent = _ACCENT_COLORS.get(self.software, _DEFAULT_ACCENT)
+
+        # History file
+        if history_file is None:
+            hist_dir = Path.home() / f".cli-anything-{self.software}"
+            hist_dir.mkdir(parents=True, exist_ok=True)
+            self.history_file = str(hist_dir / "history")
+        else:
+            self.history_file = history_file
+
+        # Detect terminal capabilities
+        self._color = self._detect_color_support()
+
+    def _detect_color_support(self) -> bool:
+        """Check if terminal supports color."""
+        if os.environ.get("NO_COLOR"):
+            return False
+        if os.environ.get("CLI_ANYTHING_NO_COLOR"):
+            return False
+        if not hasattr(sys.stdout, "isatty"):
+            return False
+        return sys.stdout.isatty()
+
+    def _c(self, code: str, text: str) -> str:
+        """Apply color code if colors are supported."""
+        if not self._color:
+            return text
+        return f"{code}{text}{_RESET}"
+
+    # ── Banner ────────────────────────────────────────────────────────
+
+    def print_banner(self):
+        """Print the startup banner with branding."""
+        import textwrap
+
+        inner = 72
+
+        def _box_line(content: str) -> str:
+            """Wrap content in box drawing, padding to inner width."""
+            pad = inner - _visible_len(content)
+            vl = self._c(_DARK_GRAY, _V_LINE)
+            return f"{vl}{content}{' ' * max(0, pad)}{vl}"
+
+        def _meta_lines(label: str, value: str) -> list[str]:
+            """Wrap a metadata line for the banner box."""
+            icon = self._c(_MAGENTA, "◇")
+            label_text = self._c(_DARK_GRAY, label)
+            prefix = f" {icon} {label_text} "
+            available = max(12, inner - _visible_len(prefix))
+            wrapped = textwrap.wrap(
+                value,
+                width=available,
+                break_long_words=True,
+                break_on_hyphens=False,
+            ) or [""]
+            lines = [f"{prefix}{self._c(_LIGHT_GRAY, wrapped[0])}"]
+            continuation_prefix = " " * _visible_len(prefix)
+            for chunk in wrapped[1:]:
+                lines.append(f"{continuation_prefix}{self._c(_LIGHT_GRAY, chunk)}")
+            return lines
+
+        top = self._c(_DARK_GRAY, f"{_TL}{_H_LINE * inner}{_TR}")
+        bot = self._c(_DARK_GRAY, f"{_BL}{_H_LINE * inner}{_BR}")
+
+        # Title:  ◆  cli-anything · Shotcut
+        icon = self._c(_CYAN + _BOLD, "◆")
+        brand = self._c(_CYAN + _BOLD, "cli-anything")
+        dot = self._c(_DARK_GRAY, "·")
+        name = self._c(self.accent + _BOLD, self.display_name)
+        title = f" {icon}  {brand} {dot} {name}"
+
+        ver = f" {self._c(_DARK_GRAY, f'   v{self.version}')}"
+        tip = f" {self._c(_DARK_GRAY, '   Type help for commands, quit to exit')}"
+        empty = ""
+
+        meta_lines: list[str] = []
+        meta_lines.extend(_meta_lines("Install:", self.skill_install_cmd))
+        meta_lines.extend(_meta_lines("Global skill:", _display_home_path(self.global_skill_path)))
+        print(top)
+        print(_box_line(title))
+        print(_box_line(ver))
+        for line in meta_lines:
+            print(_box_line(line))
+        print(_box_line(empty))
+        print(_box_line(tip))
+        print(bot)
+        print()
+
+    # ── Prompt ────────────────────────────────────────────────────────
+
+    def prompt(self, project_name: str = "", modified: bool = False,
+               context: str = "") -> str:
+        """Build a styled prompt string for prompt_toolkit or input().
+
+        Args:
+            project_name: Current project name (empty if none open).
+            modified: Whether the project has unsaved changes.
+            context: Optional extra context to show in prompt.
+
+        Returns:
+            Formatted prompt string.
+        """
+        parts = []
+
+        # Icon
+        if self._color:
+            parts.append(f"{_CYAN}◆{_RESET} ")
+        else:
+            parts.append("> ")
+
+        # Software name
+        parts.append(self._c(self.accent + _BOLD, self.software))
+
+        # Project context
+        if project_name or context:
+            ctx = context or project_name
+            mod = "*" if modified else ""
+            parts.append(f" {self._c(_DARK_GRAY, '[')}")
+            parts.append(self._c(_LIGHT_GRAY, f"{ctx}{mod}"))
+            parts.append(self._c(_DARK_GRAY, ']'))
+
+        parts.append(self._c(_GRAY, " ❯ "))
+
+        return "".join(parts)
+
+    def prompt_tokens(self, project_name: str = "", modified: bool = False,
+                      context: str = ""):
+        """Build prompt_toolkit formatted text tokens for the prompt.
+
+        Use with prompt_toolkit's FormattedText for proper ANSI handling.
+
+        Returns:
+            list of (style, text) tuples for prompt_toolkit.
+        """
+        accent_hex = _ANSI_256_TO_HEX.get(self.accent, "#5fafff")
+        tokens = []
+
+        tokens.append(("class:icon", "◆ "))
+        tokens.append(("class:software", self.software))
+
+        if project_name or context:
+            ctx = context or project_name
+            mod = "*" if modified else ""
+            tokens.append(("class:bracket", " ["))
+            tokens.append(("class:context", f"{ctx}{mod}"))
+            tokens.append(("class:bracket", "]"))
+
+        tokens.append(("class:arrow", " ❯ "))
+
+        return tokens
+
+    def get_prompt_style(self):
+        """Get a prompt_toolkit Style object matching the skin.
+
+        Returns:
+            prompt_toolkit.styles.Style
+        """
+        try:
+            from prompt_toolkit.styles import Style
+        except ImportError:
+            return None
+
+        accent_hex = _ANSI_256_TO_HEX.get(self.accent, "#5fafff")
+
+        return Style.from_dict({
+            "icon": "#5fdfdf bold",     # cyan brand color
+            "software": f"{accent_hex} bold",
+            "bracket": "#585858",
+            "context": "#bcbcbc",
+            "arrow": "#808080",
+            # Completion menu
+            "completion-menu.completion": "bg:#303030 #bcbcbc",
+            "completion-menu.completion.current": f"bg:{accent_hex} #000000",
+            "completion-menu.meta.completion": "bg:#303030 #808080",
+            "completion-menu.meta.completion.current": f"bg:{accent_hex} #000000",
+            # Auto-suggest
+            "auto-suggest": "#585858",
+            # Bottom toolbar
+            "bottom-toolbar": "bg:#1c1c1c #808080",
+            "bottom-toolbar.text": "#808080",
+        })
+
+    # ── Messages ──────────────────────────────────────────────────────
+
+    def success(self, message: str):
+        """Print a success message with green checkmark."""
+        icon = self._c(_GREEN + _BOLD, "✓")
+        print(f"  {icon} {self._c(_GREEN, message)}")
+
+    def error(self, message: str):
+        """Print an error message with red cross."""
+        icon = self._c(_RED + _BOLD, "✗")
+        print(f"  {icon} {self._c(_RED, message)}", file=sys.stderr)
+
+    def warning(self, message: str):
+        """Print a warning message with yellow triangle."""
+        icon = self._c(_YELLOW + _BOLD, "⚠")
+        print(f"  {icon} {self._c(_YELLOW, message)}")
+
+    def info(self, message: str):
+        """Print an info message with blue dot."""
+        icon = self._c(_BLUE, "●")
+        print(f"  {icon} {self._c(_LIGHT_GRAY, message)}")
+
+    def hint(self, message: str):
+        """Print a subtle hint message."""
+        print(f"  {self._c(_DARK_GRAY, message)}")
+
+    def section(self, title: str):
+        """Print a section header."""
+        print()
+        print(f"  {self._c(self.accent + _BOLD, title)}")
+        print(f"  {self._c(_DARK_GRAY, _H_LINE * len(title))}")
+
+    # ── Status display ────────────────────────────────────────────────
+
+    def status(self, label: str, value: str):
+        """Print a key-value status line."""
+        lbl = self._c(_GRAY, f"  {label}:")
+        val = self._c(_WHITE, f" {value}")
+        print(f"{lbl}{val}")
+
+    def status_block(self, items: dict[str, str], title: str = ""):
+        """Print a block of status key-value pairs.
+
+        Args:
+            items: Dict of label -> value pairs.
+            title: Optional title for the block.
+        """
+        if title:
+            self.section(title)
+
+        max_key = max(len(k) for k in items) if items else 0
+        for label, value in items.items():
+            lbl = self._c(_GRAY, f"  {label:<{max_key}}")
+            val = self._c(_WHITE, f"  {value}")
+            print(f"{lbl}{val}")
+
+    def progress(self, current: int, total: int, label: str = ""):
+        """Print a simple progress indicator.
+
+        Args:
+            current: Current step number.
+            total: Total number of steps.
+            label: Optional label for the progress.
+        """
+        pct = int(current / total * 100) if total > 0 else 0
+        bar_width = 20
+        filled = int(bar_width * current / total) if total > 0 else 0
+        bar = "█" * filled + "░" * (bar_width - filled)
+        text = f"  {self._c(_CYAN, bar)} {self._c(_GRAY, f'{pct:3d}%')}"
+        if label:
+            text += f" {self._c(_LIGHT_GRAY, label)}"
+        print(text)
+
+    # ── Table display ─────────────────────────────────────────────────
+
+    def table(self, headers: list[str], rows: list[list[str]],
+              max_col_width: int = 40):
+        """Print a formatted table with box-drawing characters.
+
+        Args:
+            headers: Column header strings.
+            rows: List of rows, each a list of cell strings.
+            max_col_width: Maximum column width before truncation.
+        """
+        if not headers:
+            return
+
+        # Calculate column widths
+        col_widths = [min(len(h), max_col_width) for h in headers]
+        for row in rows:
+            for i, cell in enumerate(row):
+                if i < len(col_widths):
+                    col_widths[i] = min(
+                        max(col_widths[i], len(str(cell))), max_col_width
+                    )
+
+        def pad(text: str, width: int) -> str:
+            t = str(text)[:width]
+            return t + " " * (width - len(t))
+
+        # Header
+        header_cells = [
+            self._c(_CYAN + _BOLD, pad(h, col_widths[i]))
+            for i, h in enumerate(headers)
+        ]
+        sep = self._c(_DARK_GRAY, f" {_V_LINE} ")
+        header_line = f"  {sep.join(header_cells)}"
+        print(header_line)
+
+        # Separator
+        sep_parts = [self._c(_DARK_GRAY, _H_LINE * w) for w in col_widths]
+        sep_line = self._c(_DARK_GRAY, f"  {'───'.join([_H_LINE * w for w in col_widths])}")
+        print(sep_line)
+
+        # Rows
+        for row in rows:
+            cells = []
+            for i, cell in enumerate(row):
+                if i < len(col_widths):
+                    cells.append(self._c(_LIGHT_GRAY, pad(str(cell), col_widths[i])))
+            row_sep = self._c(_DARK_GRAY, f" {_V_LINE} ")
+            print(f"  {row_sep.join(cells)}")
+
+    # ── Help display ──────────────────────────────────────────────────
+
+    def help(self, commands: dict[str, str]):
+        """Print a formatted help listing.
+
+        Args:
+            commands: Dict of command -> description pairs.
+        """
+        self.section("Commands")
+        max_cmd = max(len(c) for c in commands) if commands else 0
+        for cmd, desc in commands.items():
+            cmd_styled = self._c(self.accent, f"  {cmd:<{max_cmd}}")
+            desc_styled = self._c(_GRAY, f"  {desc}")
+            print(f"{cmd_styled}{desc_styled}")
+        print()
+
+    # ── Goodbye ───────────────────────────────────────────────────────
+
+    def print_goodbye(self):
+        """Print a styled goodbye message."""
+        print(f"\n  {_ICON_SMALL} {self._c(_GRAY, 'Goodbye!')}\n")
+
+    # ── Prompt toolkit session factory ────────────────────────────────
+
+    def create_prompt_session(self):
+        """Create a prompt_toolkit PromptSession with skin styling.
+
+        Returns:
+            A configured PromptSession, or None if prompt_toolkit unavailable.
+        """
+        try:
+            from prompt_toolkit import PromptSession
+            from prompt_toolkit.history import FileHistory
+            from prompt_toolkit.auto_suggest import AutoSuggestFromHistory
+            from prompt_toolkit.formatted_text import FormattedText
+
+            style = self.get_prompt_style()
+
+            session = PromptSession(
+                history=FileHistory(self.history_file),
+                auto_suggest=AutoSuggestFromHistory(),
+                style=style,
+                enable_history_search=True,
+            )
+            return session
+        except ImportError:
+            return None
+
+    def get_input(self, pt_session, project_name: str = "",
+                  modified: bool = False, context: str = "") -> str:
+        """Get input from user using prompt_toolkit or fallback.
+
+        Args:
+            pt_session: A prompt_toolkit PromptSession (or None).
+            project_name: Current project name.
+            modified: Whether project has unsaved changes.
+            context: Optional context string.
+
+        Returns:
+            User input string (stripped).
+        """
+        if pt_session is not None:
+            from prompt_toolkit.formatted_text import FormattedText
+            tokens = self.prompt_tokens(project_name, modified, context)
+            return pt_session.prompt(FormattedText(tokens)).strip()
+        else:
+            raw_prompt = self.prompt(project_name, modified, context)
+            return input(raw_prompt).strip()
+
+    # ── Toolbar builder ───────────────────────────────────────────────
+
+    def bottom_toolbar(self, items: dict[str, str]):
+        """Create a bottom toolbar callback for prompt_toolkit.
+
+        Args:
+            items: Dict of label -> value pairs to show in toolbar.
+
+        Returns:
+            A callable that returns FormattedText for the toolbar.
+        """
+        def toolbar():
+            from prompt_toolkit.formatted_text import FormattedText
+            parts = []
+            for i, (k, v) in enumerate(items.items()):
+                if i > 0:
+                    parts.append(("class:bottom-toolbar.text", "  │  "))
+                parts.append(("class:bottom-toolbar.text", f" {k}: "))
+                parts.append(("class:bottom-toolbar", v))
+            return FormattedText(parts)
+        return toolbar
+
+
+# ── ANSI 256-color to hex mapping (for prompt_toolkit styles) ─────────
+
+_ANSI_256_TO_HEX = {
+    "\033[38;5;33m":  "#0087ff",  # audacity navy blue
+    "\033[38;5;35m":  "#00af5f",  # shotcut teal
+    "\033[38;5;39m":  "#00afff",  # inkscape bright blue
+    "\033[38;5;40m":  "#00d700",  # libreoffice green
+    "\033[38;5;55m":  "#5f00af",  # obs purple
+    "\033[38;5;69m":  "#5f87ff",  # kdenlive slate blue
+    "\033[38;5;75m":  "#5fafff",  # default sky blue
+    "\033[38;5;80m":  "#5fd7d7",  # brand cyan
+    "\033[38;5;208m": "#ff8700",  # blender deep orange
+    "\033[38;5;214m": "#ffaf00",  # gimp warm orange
+}

tigris/agent-harness/cli_anything/tigris/utils/tigris_backend.py

@@ -0,0 +1,315 @@
+"""Subprocess wrapper around the official Tigris CLI (`tigris`).
+
+This harness shells out to the Tigris CLI rather than reimplementing the S3
+protocol, so:
+
+  * snapshots, IAM, access-keys, and organization primitives — features
+    unique to Tigris — are surfaced for agents.
+  * the harness inherits new commands automatically as the upstream CLI ships
+    them.
+  * authentication uses Tigris's OAuth flow (`tigris login`) by default.
+
+Install the underlying CLI with one of:
+
+    npm install -g @tigrisdata/cli
+    brew install tigrisdata/tap/tigris
+
+Then `tigris login` once, and this harness can call any operation as the
+authenticated user.
+"""
+
+import json as json_mod
+import os
+import shutil
+import subprocess
+from typing import Any
+
+
+class TigrisCliError(RuntimeError):
+    """Raised when the tigris CLI returns a non-zero exit code."""
+
+
+def _path_to_t3(path: str) -> str:
+    """Normalize 'tigris://...', 't3://...', or bare 'bucket/key' to t3://..."""
+    if path.startswith("t3://"):
+        return path
+    if path.startswith("tigris://"):
+        return "t3://" + path[len("tigris://"):]
+    return "t3://" + path.lstrip("/")
+
+
+class TigrisBackend:
+    """Subprocess wrapper for the `tigris` CLI."""
+
+    def __init__(self, cli_path: str = "tigris", access_key: str | None = None,
+                 secret_key: str | None = None):
+        """Resolve the CLI binary; verify it's on PATH.
+
+        Args:
+            cli_path: Name or absolute path of the tigris binary. Default
+                `tigris`; the binary's alias `t3` also works.
+            access_key, secret_key: Optional explicit credentials. When set,
+                exported into env for child processes so commands that honor
+                AWS_* env vars pick them up. Most users should `tigris login`
+                instead.
+        """
+        resolved = shutil.which(cli_path)
+        if not resolved:
+            raise TigrisCliError(
+                f"`{cli_path}` not found on PATH. Install with "
+                "`npm install -g @tigrisdata/cli` or "
+                "`brew install tigrisdata/tap/tigris`, then `tigris login`."
+            )
+        self.cli_path = resolved
+        self._extra_env: dict[str, str] = {}
+        if access_key:
+            self._extra_env["TIGRIS_STORAGE_ACCESS_KEY_ID"] = access_key
+            self._extra_env["AWS_ACCESS_KEY_ID"] = access_key
+        if secret_key:
+            self._extra_env["TIGRIS_STORAGE_SECRET_ACCESS_KEY"] = secret_key
+            self._extra_env["AWS_SECRET_ACCESS_KEY"] = secret_key
+
+    # ── core invoker ──────────────────────────────────────────────────
+
+    def _run(
+        self,
+        args: list[str],
+        json: bool = True,
+        check: bool = True,
+        capture: bool = True,
+    ) -> Any:
+        """Run `tigris <args>` and return parsed JSON when possible.
+
+        Args:
+            args: CLI subcommand + flags, excluding the binary name itself.
+            json: If True, append `--format json` and parse stdout as JSON.
+                Falls back to raw stdout on parse failure.
+            check: If True, raise TigrisCliError on non-zero exit.
+            capture: If True, capture stdout/stderr; if False, stream to the
+                caller's TTY (used for `login` / interactive flows).
+
+        Returns:
+            Parsed JSON object on success when json=True, raw string when
+            json=False or parse fails, or None when capture=False.
+        """
+        cmd = [self.cli_path] + list(args)
+        if json:
+            # Append at the end so it doesn't fight with subcommand parsing.
+            cmd.extend(["--format", "json"])
+
+        env = {**os.environ, **self._extra_env}
+        if not capture:
+            result = subprocess.run(cmd, env=env, check=False)
+            if check and result.returncode != 0:
+                raise TigrisCliError(
+                    f"`{' '.join(cmd)}` exited with code {result.returncode}"
+                )
+            return None
+
+        result = subprocess.run(
+            cmd, env=env, check=False, capture_output=True, text=True
+        )
+        if check and result.returncode != 0:
+            stderr = result.stderr.strip() or result.stdout.strip()
+            raise TigrisCliError(
+                f"`{' '.join(cmd)}` failed: {stderr or 'no output'}"
+            )
+        if json:
+            stdout = result.stdout.strip()
+            if not stdout:
+                return None
+            try:
+                return json_mod.loads(stdout)
+            except json_mod.JSONDecodeError:
+                # Not all commands support --format json; return raw text.
+                return stdout
+        return result.stdout
+
+    # ── version / auth ────────────────────────────────────────────────
+
+    def version(self) -> str:
+        """Return the tigris CLI version string."""
+        return (self._run(["--version"], json=False) or "").strip()
+
+    def whoami(self) -> Any:
+        """Return the currently authenticated user/org."""
+        return self._run(["whoami"])
+
+    def login(self) -> None:
+        """Trigger the interactive `tigris login` flow (browser OAuth).
+
+        Streams to the caller's TTY; no return value. Use --json=False since
+        the flow prompts and prints progress.
+        """
+        self._run(["login"], json=False, capture=False)
+
+    def logout(self) -> None:
+        """Log out of the currently authenticated session."""
+        self._run(["logout"], json=False, capture=False)
+
+    # ── buckets ───────────────────────────────────────────────────────
+
+    def list_buckets(self) -> Any:
+        return self._run(["buckets", "list"])
+
+    def create_bucket(self, name: str) -> Any:
+        return self._run(["buckets", "create", name])
+
+    def delete_bucket(self, name: str, yes: bool = False) -> Any:
+        args = ["buckets", "delete", name]
+        if yes:
+            args.append("--yes")
+        return self._run(args)
+
+    def head_bucket(self, name: str) -> Any:
+        """Get bucket info via `buckets get` (falls back to `stat`)."""
+        # The CLI exposes `tigris buckets get <name>` as the head equivalent.
+        return self._run(["buckets", "get", name])
+
+    # ── objects ───────────────────────────────────────────────────────
+
+    def list_objects(
+        self,
+        bucket: str,
+        prefix: str | None = None,
+        limit: int | None = None,
+    ) -> Any:
+        """`tigris ls <bucket[/prefix]>` — limit handled client-side if set."""
+        path = bucket if not prefix else f"{bucket}/{prefix}".rstrip("/")
+        result = self._run(["ls", _path_to_t3(path)])
+        if isinstance(result, list) and limit:
+            return result[:limit]
+        return result
+
+    def cp(self, src: str, dst: str, recursive: bool = False) -> Any:
+        """Server-side or local↔remote copy.
+
+        `src` and `dst` may be local paths or t3:// / tigris:// URIs.
+        At least one side must be remote (enforced by the underlying CLI).
+        """
+        src_norm = _path_to_t3(src) if src.startswith(("t3://", "tigris://")) else src
+        dst_norm = _path_to_t3(dst) if dst.startswith(("t3://", "tigris://")) else dst
+        args = ["cp", src_norm, dst_norm]
+        if recursive:
+            args.append("--recursive")
+        return self._run(args, json=False)
+
+    def put_object_from_file(self, bucket: str, key: str, file_path: str) -> Any:
+        """Upload a local file via `tigris cp <file> t3://bucket/key`."""
+        return self.cp(file_path, f"t3://{bucket}/{key}")
+
+    def put_object_inline(self, bucket: str, key: str, text: str) -> Any:
+        """Upload inline text by staging to a tempfile and `tigris cp`-ing it."""
+        import tempfile
+        with tempfile.NamedTemporaryFile(
+            mode="w", encoding="utf-8", suffix=".txt", delete=False
+        ) as f:
+            f.write(text)
+            tmp_path = f.name
+        try:
+            return self.cp(tmp_path, f"t3://{bucket}/{key}")
+        finally:
+            try:
+                os.unlink(tmp_path)
+            except OSError:
+                pass
+
+    def get_object_to_file(self, bucket: str, key: str, file_path: str) -> Any:
+        """Download an object via `tigris cp t3://bucket/key <file>`."""
+        return self.cp(f"t3://{bucket}/{key}", file_path)
+
+    def delete_object(self, bucket: str, key: str) -> Any:
+        return self._run(["rm", _path_to_t3(f"{bucket}/{key}"), "--yes"], json=False)
+
+    def head_object(self, bucket: str, key: str) -> Any:
+        """`tigris stat t3://bucket/key`."""
+        return self._run(["stat", _path_to_t3(f"{bucket}/{key}")])
+
+    # ── presign ───────────────────────────────────────────────────────
+
+    def presign(
+        self,
+        bucket: str,
+        key: str,
+        method: str = "get",
+        expires_in: int = 3600,
+        access_key: str | None = None,
+    ) -> str:
+        """Generate a presigned URL via `tigris presign`."""
+        args = [
+            "presign",
+            _path_to_t3(f"{bucket}/{key}"),
+            "--method", method,
+            "--expires-in", str(expires_in),
+        ]
+        if access_key:
+            args.extend(["--access-key", access_key])
+        result = self._run(args)
+        # The CLI returns either {"url": "..."} (json mode) or the bare URL.
+        if isinstance(result, dict) and "url" in result:
+            return result["url"]
+        if isinstance(result, str):
+            return result.strip()
+        return str(result)
+
+    # ── snapshots ─────────────────────────────────────────────────────
+
+    def list_snapshots(self, bucket: str) -> Any:
+        return self._run(["snapshots", "list", bucket])
+
+    def take_snapshot(self, bucket: str, name: str | None = None) -> Any:
+        args = ["snapshots", "take", bucket]
+        if name:
+            args.extend(["--name", name])
+        return self._run(args)
+
+    # ── access keys ───────────────────────────────────────────────────
+
+    def list_access_keys(self) -> Any:
+        return self._run(["access-keys", "list"])
+
+    def create_access_key(self, name: str) -> Any:
+        return self._run(["access-keys", "create", name])
+
+    def get_access_key(self, key_id: str) -> Any:
+        return self._run(["access-keys", "get", key_id])
+
+    def delete_access_key(self, key_id: str, yes: bool = False) -> Any:
+        args = ["access-keys", "delete", key_id]
+        if yes:
+            args.append("--yes")
+        return self._run(args)
+
+    def assign_access_key(
+        self, key_id: str, bucket: str, role: str
+    ) -> Any:
+        """`tigris access-keys assign <id> --bucket <b> --role <r>`."""
+        return self._run([
+            "access-keys", "assign", key_id,
+            "--bucket", bucket, "--role", role,
+        ])
+
+    def rotate_access_key(self, key_id: str, yes: bool = False) -> Any:
+        args = ["access-keys", "rotate", key_id]
+        if yes:
+            args.append("--yes")
+        return self._run(args)
+
+    # ── IAM ───────────────────────────────────────────────────────────
+
+    def list_iam_policies(self) -> Any:
+        return self._run(["iam", "policies", "list"])
+
+    def create_iam_policy(self, name: str, document_path: str) -> Any:
+        return self._run([
+            "iam", "policies", "create", name,
+            "--document", document_path,
+        ])
+
+    def list_iam_users(self) -> Any:
+        return self._run(["iam", "users", "list"])
+
+    def invite_iam_user(self, email: str, role: str = "member") -> Any:
+        return self._run([
+            "iam", "users", "invite", email, "--role", role,
+        ])

tigris/agent-harness/setup.py

@@ -0,0 +1,56 @@
+"""Setup for cli-anything-tigris — CLI harness for the Tigris object storage CLI.
+
+This harness shells out to the official `tigris` CLI rather than reimplementing
+the S3 protocol, so it surfaces every Tigris primitive (snapshots, IAM,
+scoped access keys, OAuth login) — not just generic S3 ops.
+
+Install the underlying CLI first with one of:
+
+    npm install -g @tigrisdata/cli
+    brew install tigrisdata/tap/tigris
+"""
+
+from setuptools import setup, find_namespace_packages
+
+setup(
+    name="cli-anything-tigris",
+    version="1.0.0",
+    author="cli-anything contributors",
+    author_email="",
+    description="CLI-Anything harness wrapping the official Tigris CLI (object storage, snapshots, IAM, presign)",
+    url="https://github.com/HKUDS/CLI-Anything",
+    packages=find_namespace_packages(include=["cli_anything.*"]),
+    classifiers=[
+        "Development Status :: 4 - Beta",
+        "Intended Audience :: Developers",
+        "Topic :: Software Development :: Libraries :: Python Modules",
+        "License :: OSI Approved :: MIT License",
+        "Programming Language :: Python :: 3",
+        "Programming Language :: Python :: 3.10",
+        "Programming Language :: Python :: 3.11",
+        "Programming Language :: Python :: 3.12",
+    ],
+    python_requires=">=3.10",
+    install_requires=[
+        "click>=8.0.0",
+        "prompt-toolkit>=3.0.0",
+        # NB: no boto3. This harness shells out to the `tigris` CLI, which is
+        # an external runtime dependency installed via npm or brew.
+    ],
+    extras_require={
+        "dev": [
+            "pytest>=7.0.0",
+            "pytest-cov>=4.0.0",
+        ],
+    },
+    entry_points={
+        "console_scripts": [
+            "cli-anything-tigris=cli_anything.tigris.tigris_cli:main",
+        ],
+    },
+    package_data={
+        "cli_anything.tigris": ["skills/*.md"],
+    },
+    include_package_data=True,
+    zip_safe=False,
+)