diff --git a/.codex-plugin/plugin.json b/.codex-plugin/plugin.json index ba2d4f8..1300911 100644 --- a/.codex-plugin/plugin.json +++ b/.codex-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "codex-settings", - "version": "1.2.0", + "version": "1.2.1", "description": "Reusable Codex skills and workflows from feiskyer/codex-settings.", "author": { "name": "feiskyer", diff --git a/skills/atlas-image-skill/SKILL.md b/skills/atlas-image-skill/SKILL.md new file mode 100644 index 0000000..2e7cd8a --- /dev/null +++ b/skills/atlas-image-skill/SKILL.md @@ -0,0 +1,39 @@ +--- +name: atlas-image-skill +description: 'Generate images with the Atlas Cloud API. Use when the user asks for Atlas Cloud image generation, an Atlas image model, or an image from a text prompt through Atlas Cloud.' +--- + +# Atlas Image Skill + +Generate an image from a text prompt through the bundled Atlas Cloud helper. + +## Resolve the skill directory + +Resolve the absolute directory containing this `SKILL.md` before running a command and refer to it as ``. Keep output paths relative to the user's working directory unless the user requests another location. + +## Requirements + +- Export `ATLASCLOUD_API_KEY` before running the helper. `ATLAS_CLOUD_API_KEY` is also accepted for compatibility. +- Use Python 3.9 or newer. The helper uses only the Python standard library. + +## Generate an image + +Confirm the prompt and output filename before submitting a generation request. Then run: + +```bash +python3 "/atlas_image.py" \ + --prompt "A quiet observatory beneath an aurora" \ + --output "observatory.png" +``` + +The helper defaults to `google/nano-banana-2-lite/text-to-image-developer`. Use `--model` to select another current Atlas text-to-image model. Optional fields include `--aspect-ratio` and `--thinking-level`. + +The API is asynchronous. The helper submits exactly one generation request, polls the returned prediction with bounded GET requests, downloads the first HTTPS output, and creates output parent directories automatically. + +## Safety and failures + +- Never print or persist the API key. +- Validate the prompt, API base, and local output path before calling the API. +- Never retry the generation POST because another request may incur another charge. +- Retry only transient prediction GET failures, with bounded exponential backoff. +- Do not claim success unless the generated image was saved locally. diff --git a/skills/atlas-image-skill/agents/openai.yaml b/skills/atlas-image-skill/agents/openai.yaml new file mode 100644 index 0000000..92d7f36 --- /dev/null +++ b/skills/atlas-image-skill/agents/openai.yaml @@ -0,0 +1,4 @@ +interface: + display_name: "Atlas Cloud Image" + short_description: "Generate images from text with the Atlas Cloud API" + default_prompt: "Use $atlas-image-skill to generate an image with Atlas Cloud and save it locally." diff --git a/skills/atlas-image-skill/atlas_image.py b/skills/atlas-image-skill/atlas_image.py new file mode 100755 index 0000000..998abb5 --- /dev/null +++ b/skills/atlas-image-skill/atlas_image.py @@ -0,0 +1,352 @@ +#!/usr/bin/env python3 +"""Generate images from text with the Atlas Cloud API.""" + +import argparse +import json +import os +import sys +import time +import uuid +from pathlib import Path +from urllib.error import HTTPError, URLError +from urllib.parse import quote, urlparse +from urllib.request import Request, urlopen + +DEFAULT_API_BASE = "https://api.atlascloud.ai" +DEFAULT_MODEL = "google/nano-banana-2-lite/text-to-image-developer" +USER_AGENT = "codex-settings-atlas-image-skill/1.0" +SUPPORTED_ASPECT_RATIOS = ( + "auto", + "1:1", + "3:2", + "2:3", + "3:4", + "4:3", + "4:5", + "5:4", + "9:16", + "16:9", + "21:9", + "4:1", + "1:4", + "8:1", + "1:8", +) +SUPPORTED_THINKING_LEVELS = ("default", "high", "minimal") +PENDING_STATUSES = {"created", "starting", "processing"} +REQUEST_TIMEOUT_SECONDS = 60 +DEFAULT_POLL_INTERVAL_SECONDS = 2.0 +DEFAULT_POLL_TIMEOUT_SECONDS = 300.0 +MAX_JSON_BYTES = 1024 * 1024 +MAX_DOWNLOAD_BYTES = 50 * 1024 * 1024 +GET_MAX_RETRIES = 3 + + +def positive_float(value): + """Parse a positive floating-point number for argparse.""" + parsed = float(value) + if parsed <= 0: + raise argparse.ArgumentTypeError("must be greater than zero") + return parsed + + +def normalize_api_base(value): + """Validate and normalize an HTTPS API base URL.""" + parsed = urlparse(value) + if parsed.scheme != "https" or not parsed.netloc: + raise ValueError("--api-base must be an HTTPS URL") + if parsed.params or parsed.query or parsed.fragment: + raise ValueError("--api-base must not contain parameters, a query, or a fragment") + return value.rstrip("/") + + +def build_parser(): + parser = argparse.ArgumentParser( + description="Generate images from text with the Atlas Cloud API" + ) + parser.add_argument("--prompt", required=True, help="Text prompt for image generation") + parser.add_argument( + "--output", + default=None, + help="Output filename (default: atlas-image-.png)", + ) + parser.add_argument( + "--model", + default=DEFAULT_MODEL, + help=f"Atlas image model (default: {DEFAULT_MODEL})", + ) + parser.add_argument( + "--aspect-ratio", + choices=SUPPORTED_ASPECT_RATIOS, + default="auto", + help="Requested aspect ratio (default: auto)", + ) + parser.add_argument( + "--thinking-level", + choices=SUPPORTED_THINKING_LEVELS, + default="default", + help="Model thinking level (default: default)", + ) + parser.add_argument( + "--poll-interval", + type=positive_float, + default=DEFAULT_POLL_INTERVAL_SECONDS, + help="Seconds between prediction checks (default: 2)", + ) + parser.add_argument( + "--timeout", + type=positive_float, + default=DEFAULT_POLL_TIMEOUT_SECONDS, + help="Maximum prediction wait in seconds (default: 300)", + ) + parser.add_argument( + "--api-base", + default=None, + help="Atlas Cloud API base URL (default: ATLASCLOUD_API_BASE or api.atlascloud.ai)", + ) + return parser + + +def parse_args(argv=None): + parser = build_parser() + args = parser.parse_args(argv) + args.prompt = args.prompt.strip() + if not args.prompt: + parser.error("--prompt must not be empty") + if len(args.prompt) > 10000: + parser.error("--prompt must not exceed 10000 characters") + args.model = args.model.strip() + if not args.model: + parser.error("--model must not be empty") + args.output = Path(args.output or f"atlas-image-{uuid.uuid4()}.png").expanduser() + if args.output.name in {"", ".", ".."} or args.output.is_dir(): + parser.error("--output must name a file") + configured_base = ( + args.api_base + or os.getenv("ATLASCLOUD_API_BASE") + or os.getenv("ATLAS_CLOUD_API_BASE") + or DEFAULT_API_BASE + ) + try: + args.api_base = normalize_api_base(configured_base) + except ValueError as exc: + parser.error(str(exc)) + return args + + +def build_payload(args): + return { + "model": args.model, + "prompt": args.prompt, + "aspect_ratio": args.aspect_ratio, + "thinking_level": args.thinking_level, + "resolution": "1k", + "enable_sync_mode": False, + "enable_base64_output": False, + } + + +def read_json(response): + body = response.read(MAX_JSON_BYTES + 1) + if len(body) > MAX_JSON_BYTES: + raise RuntimeError("Atlas Cloud response exceeds the maximum allowed size") + try: + payload = json.loads(body.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise RuntimeError("Atlas Cloud returned an invalid JSON response") from exc + if not isinstance(payload, dict): + raise TypeError("Atlas Cloud returned an invalid response object") + return payload + + +def response_data(payload): + code = payload.get("code") + if code not in (None, 0, 200): + message = payload.get("message") or "unknown API error" + raise RuntimeError(f"Atlas Cloud request failed: {message}") + data = payload.get("data") + if not isinstance(data, dict): + raise TypeError("Atlas Cloud response did not contain prediction data") + return data + + +def api_request(request, opener=urlopen): + try: + with opener(request, timeout=REQUEST_TIMEOUT_SECONDS) as response: + return response_data(read_json(response)) + except HTTPError as exc: + raise RuntimeError(f"Atlas Cloud request failed with HTTP {exc.code}") from exc + except URLError as exc: + raise RuntimeError(f"Atlas Cloud request failed: {exc.reason}") from exc + + +def submit_generation(args, api_key, opener=urlopen): + """Submit exactly one generation POST and return its prediction data.""" + request = Request( + f"{args.api_base}/api/v1/model/generateImage", + data=json.dumps(build_payload(args)).encode("utf-8"), + headers={ + "Authorization": f"Bearer {api_key}", + "Content-Type": "application/json", + "User-Agent": USER_AGENT, + }, + method="POST", + ) + data = api_request(request, opener=opener) + prediction_id = data.get("id") + if not isinstance(prediction_id, str) or not prediction_id: + raise RuntimeError("Atlas Cloud response did not contain a prediction ID") + return data + + +def request_prediction( + api_base, + prediction_id, + api_key, + opener=urlopen, + sleep_fn=time.sleep, +): + """Fetch a prediction, retrying only transient GET failures.""" + request = Request( + f"{api_base}/api/v1/model/prediction/{quote(prediction_id, safe='')}", + headers={ + "Authorization": f"Bearer {api_key}", + "User-Agent": USER_AGENT, + }, + method="GET", + ) + for attempt in range(GET_MAX_RETRIES + 1): + try: + with opener(request, timeout=REQUEST_TIMEOUT_SECONDS) as response: + return response_data(read_json(response)) + except HTTPError as exc: + transient = exc.code == 429 or 500 <= exc.code < 600 + if not transient or attempt == GET_MAX_RETRIES: + raise RuntimeError( + f"Atlas Cloud prediction request failed with HTTP {exc.code}" + ) from exc + except URLError as exc: + if attempt == GET_MAX_RETRIES: + raise RuntimeError( + f"Atlas Cloud prediction request failed: {exc.reason}" + ) from exc + sleep_fn(2**attempt) + raise RuntimeError("Atlas Cloud prediction request exhausted retries") + + +def completed_outputs(data): + status = data.get("status") + if status == "completed": + outputs = data.get("outputs") + if not isinstance(outputs, list) or not outputs: + raise RuntimeError("Completed Atlas Cloud prediction contained no outputs") + if not all(isinstance(value, str) and value for value in outputs): + raise RuntimeError("Atlas Cloud prediction contained an invalid output") + return outputs + if status in {"failed", "timeout"}: + detail = data.get("error") or f"prediction {status}" + raise RuntimeError(f"Atlas Cloud generation failed: {detail}") + if status not in PENDING_STATUSES: + raise RuntimeError(f"Atlas Cloud returned an unknown prediction status: {status}") + return None + + +def wait_for_outputs( + initial_data, + args, + api_key, + opener=urlopen, + sleep_fn=time.sleep, + monotonic_fn=time.monotonic, +): + outputs = completed_outputs(initial_data) + if outputs: + return outputs + deadline = monotonic_fn() + args.timeout + prediction_id = initial_data["id"] + while monotonic_fn() < deadline: + data = request_prediction( + args.api_base, + prediction_id, + api_key, + opener=opener, + sleep_fn=sleep_fn, + ) + outputs = completed_outputs(data) + if outputs: + return outputs + remaining = deadline - monotonic_fn() + if remaining > 0: + sleep_fn(min(args.poll_interval, remaining)) + raise RuntimeError("Atlas Cloud generation timed out while polling") + + +def download_image(url, opener=urlopen): + parsed = urlparse(url) + if parsed.scheme != "https" or not parsed.netloc: + raise RuntimeError("Atlas Cloud returned an invalid image URL") + request = Request( + url, + headers={"Accept": "image/*", "User-Agent": USER_AGENT}, + ) + try: + with opener(request, timeout=REQUEST_TIMEOUT_SECONDS) as response: + data = response.read(MAX_DOWNLOAD_BYTES + 1) + except HTTPError as exc: + raise RuntimeError(f"Image download failed with HTTP {exc.code}") from exc + except URLError as exc: + raise RuntimeError(f"Image download failed: {exc.reason}") from exc + if len(data) > MAX_DOWNLOAD_BYTES: + raise RuntimeError("Downloaded image exceeds the maximum allowed size") + if not data: + raise RuntimeError("Downloaded image is empty") + return data + + +def save_first_output(outputs, output_path, opener=urlopen): + image_data = download_image(outputs[0], opener=opener) + output_path = Path(output_path) + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_bytes(image_data) + return str(output_path) + + +def run( + args, + api_key=None, + api_opener=urlopen, + download_opener=urlopen, + sleep_fn=time.sleep, + monotonic_fn=time.monotonic, +): + active_api_key = ( + api_key + or os.getenv("ATLASCLOUD_API_KEY") + or os.getenv("ATLAS_CLOUD_API_KEY") + ) + if not active_api_key: + raise RuntimeError("ATLASCLOUD_API_KEY is required") + initial_data = submit_generation(args, active_api_key, opener=api_opener) + outputs = wait_for_outputs( + initial_data, + args, + active_api_key, + opener=api_opener, + sleep_fn=sleep_fn, + monotonic_fn=monotonic_fn, + ) + saved_path = save_first_output(outputs, args.output, opener=download_opener) + print(f"Image saved to: {saved_path}") + return saved_path + + +def main(argv=None): + try: + return 0 if run(parse_args(argv)) else 1 + except (OSError, RuntimeError, TypeError, ValueError) as exc: + print(f"Error: {exc}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/skills/atlas-image-skill/tests/test_atlas_image.py b/skills/atlas-image-skill/tests/test_atlas_image.py new file mode 100644 index 0000000..0ba8350 --- /dev/null +++ b/skills/atlas-image-skill/tests/test_atlas_image.py @@ -0,0 +1,268 @@ +import importlib.util +import io +import json +import os +import tempfile +import unittest +from contextlib import redirect_stderr +from pathlib import Path +from unittest.mock import patch +from urllib.error import HTTPError + +SKILL_DIR = Path(__file__).resolve().parents[1] +MODULE_PATH = SKILL_DIR / "atlas_image.py" +SPEC = importlib.util.spec_from_file_location("atlas_image", MODULE_PATH) +atlas_image = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(atlas_image) + + +class FakeResponse: + def __init__(self, body): + self.body = body + + def __enter__(self): + return self + + def __exit__(self, *_args): + return False + + def read(self, size=-1): + return self.body[:size] if size >= 0 else self.body + + +class RecordingOpener: + def __init__(self, responses): + self.responses = list(responses) + self.calls = [] + + def __call__(self, request, timeout): + self.calls.append((request, timeout)) + response = self.responses.pop(0) + if isinstance(response, Exception): + raise response + return FakeResponse(response) + + +def api_response(data, code=200, message=""): + return json.dumps({"code": code, "message": message, "data": data}).encode() + + +class AtlasImageTests(unittest.TestCase): + def test_submit_uses_live_contract_and_exactly_one_post(self): + opener = RecordingOpener( + [api_response({"id": "prediction-1", "status": "created"})] + ) + args = atlas_image.parse_args( + [ + "--prompt", + "A geometric garden", + "--aspect-ratio", + "16:9", + "--thinking-level", + "high", + ] + ) + + data = atlas_image.submit_generation(args, "test-key", opener=opener) + + self.assertEqual(data["id"], "prediction-1") + self.assertEqual(len(opener.calls), 1) + request = opener.calls[0][0] + payload = json.loads(request.data) + self.assertEqual(request.get_method(), "POST") + self.assertEqual( + request.full_url, + "https://api.atlascloud.ai/api/v1/model/generateImage", + ) + self.assertEqual(request.headers["Authorization"], "Bearer test-key") + self.assertEqual( + request.get_header("User-agent"), + "codex-settings-atlas-image-skill/1.0", + ) + self.assertEqual( + payload["model"], + "google/nano-banana-2-lite/text-to-image-developer", + ) + self.assertEqual(payload["aspect_ratio"], "16:9") + self.assertEqual(payload["thinking_level"], "high") + self.assertFalse(payload["enable_sync_mode"]) + + def test_run_polls_with_get_and_saves_first_output(self): + api_opener = RecordingOpener( + [ + api_response({"id": "prediction-2", "status": "created"}), + api_response({"id": "prediction-2", "status": "processing"}), + api_response( + { + "id": "prediction-2", + "status": "completed", + "outputs": ["https://cdn.example/result.png"], + } + ), + ] + ) + downloader = RecordingOpener([b"generated-image"]) + clock = iter([0.0, 0.0, 0.1, 0.1]) + with tempfile.TemporaryDirectory() as temp_dir: + output = Path(temp_dir) / "nested" / "result.png" + args = atlas_image.parse_args( + ["--prompt", "A quiet observatory", "--output", str(output)] + ) + + saved = atlas_image.run( + args, + api_key="test-key", + api_opener=api_opener, + download_opener=downloader, + sleep_fn=lambda _seconds: None, + monotonic_fn=lambda: next(clock), + ) + + self.assertEqual(saved, str(output)) + self.assertEqual(output.read_bytes(), b"generated-image") + methods = [call[0].get_method() for call in api_opener.calls] + self.assertEqual(methods, ["POST", "GET", "GET"]) + + def test_completed_submit_does_not_poll(self): + api_opener = RecordingOpener( + [ + api_response( + { + "id": "prediction-3", + "status": "completed", + "outputs": ["https://cdn.example/result.png"], + } + ) + ] + ) + downloader = RecordingOpener([b"generated-image"]) + with tempfile.TemporaryDirectory() as temp_dir: + args = atlas_image.parse_args( + [ + "--prompt", + "A lighthouse", + "--output", + str(Path(temp_dir) / "result.png"), + ] + ) + atlas_image.run( + args, + api_key="test-key", + api_opener=api_opener, + download_opener=downloader, + ) + self.assertEqual(len(api_opener.calls), 1) + self.assertEqual(api_opener.calls[0][0].get_method(), "POST") + + def test_transient_get_is_retried_but_post_is_not(self): + error = HTTPError("https://example", 503, "Unavailable", {}, None) + opener = RecordingOpener( + [ + error, + api_response( + { + "id": "prediction-4", + "status": "completed", + "outputs": ["https://cdn.example/result.png"], + } + ), + ] + ) + sleeps = [] + + data = atlas_image.request_prediction( + "https://api.atlascloud.ai", + "prediction-4", + "test-key", + opener=opener, + sleep_fn=sleeps.append, + ) + + self.assertEqual(data["status"], "completed") + self.assertEqual(len(opener.calls), 2) + self.assertEqual(sleeps, [1]) + self.assertEqual( + opener.calls[0][0].get_header("User-agent"), + "codex-settings-atlas-image-skill/1.0", + ) + + def test_invalid_paid_inputs_are_rejected_locally(self): + invalid_arguments = [ + ["--prompt", " "], + ["--prompt", "test", "--api-base", "http://api.example"], + ["--prompt", "test", "--timeout", "0"], + ] + for arguments in invalid_arguments: + with ( + self.subTest(arguments=arguments), + redirect_stderr(io.StringIO()), + self.assertRaises(SystemExit), + ): + atlas_image.parse_args(arguments) + + def test_failed_prediction_reports_api_error(self): + args = atlas_image.parse_args(["--prompt", "A lighthouse"]) + with self.assertRaisesRegex(RuntimeError, "content policy"): + atlas_image.wait_for_outputs( + { + "id": "prediction-5", + "status": "failed", + "error": "content policy", + }, + args, + "test-key", + ) + + def test_polling_timeout_is_bounded(self): + args = atlas_image.parse_args( + ["--prompt", "A lighthouse", "--timeout", "1"] + ) + clock = iter([0.0, 1.0]) + with self.assertRaisesRegex(RuntimeError, "timed out"): + atlas_image.wait_for_outputs( + {"id": "prediction-6", "status": "created"}, + args, + "test-key", + monotonic_fn=lambda: next(clock), + ) + + def test_compatibility_api_key_is_used(self): + api_opener = RecordingOpener( + [ + api_response( + { + "id": "prediction-7", + "status": "completed", + "outputs": ["https://cdn.example/result.png"], + } + ) + ] + ) + downloader = RecordingOpener([b"generated-image"]) + with tempfile.TemporaryDirectory() as temp_dir: + args = atlas_image.parse_args( + [ + "--prompt", + "A lighthouse", + "--output", + str(Path(temp_dir) / "result.png"), + ] + ) + with patch.dict( + os.environ, + {"ATLAS_CLOUD_API_KEY": "compat-key"}, + clear=True, + ): + atlas_image.run( + args, + api_opener=api_opener, + download_opener=downloader, + ) + self.assertEqual( + api_opener.calls[0][0].headers["Authorization"], + "Bearer compat-key", + ) + + +if __name__ == "__main__": + unittest.main()