diff --git a/.gitignore b/.gitignore index 7f20384c..0f2dca62 100644 --- a/.gitignore +++ b/.gitignore @@ -111,6 +111,13 @@ src/aipass/*/apps/integrations/** !src/aipass/spawn/templates/builder/docs.local/** !src/aipass/spawn/templates/builder/DASHBOARD.local.json +# Commons artifacts subsystem — real source code (craft/trade/capsule), NOT a +# runtime dir. Collides with the blanket `artifacts/` ignore (line 49); *.py-only +# negation keeps the logs/ + __pycache__/ subdirs ignored. Without this the +# tracked test_artifacts.py imports a module absent from CI -> ImportError. +!src/aipass/commons/apps/handlers/artifacts/ +!src/aipass/commons/apps/handlers/artifacts/*.py + # CI artifacts windows-pytest-results/ diff --git a/src/aipass/commons/apps/handlers/artifacts/__init__.py b/src/aipass/commons/apps/handlers/artifacts/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/src/aipass/commons/apps/handlers/artifacts/artifact_ops.py b/src/aipass/commons/apps/handlers/artifacts/artifact_ops.py new file mode 100644 index 00000000..f68d9607 --- /dev/null +++ b/src/aipass/commons/apps/handlers/artifacts/artifact_ops.py @@ -0,0 +1,540 @@ +# =================== AIPass ==================== +# Name: artifact_ops.py +# Description: Artifact Operations Handler +# Version: 1.0.0 +# Created: 2026-03-07 +# Modified: 2026-03-07 +# ============================================= + +""" +Artifact Operations Handler + +Implementation logic for artifact workflows: craft, list, inspect, +birth certificates, and joint artifact collaboration. +Returns dicts for module display layer. +""" + +import json +import os +from typing import List, Optional +from datetime import datetime, timezone, timedelta + +from aipass.prax.apps.modules.logger import system_logger as logger + +from aipass.commons.apps.handlers.database.db import get_db, close_db +from aipass.commons.apps.handlers.json import json_handler + + +# Constants +def _find_registry_path() -> str: + """Walk up from __file__ to find AIPASS_REGISTRY.json in the project root.""" + current = os.path.dirname(os.path.abspath(__file__)) + for _ in range(10): + candidate = os.path.join(current, "AIPASS_REGISTRY.json") + if os.path.exists(candidate): + return candidate + current = os.path.dirname(current) + return os.path.join(os.path.expanduser("~"), "AIPASS_REGISTRY.json") + + +BRANCH_REGISTRY_PATH = _find_registry_path() + +VALID_RARITIES = ("common", "uncommon", "rare", "legendary", "unique") +VALID_TYPES = ("crafted", "found", "birth_certificate", "event", "seasonal", "joint", "system") + +RARITY_COLORS = { + "common": "white", + "uncommon": "green", + "rare": "blue", + "legendary": "yellow", + "unique": "magenta", +} + + +# ============================================================================= +# HELPER FUNCTIONS +# ============================================================================= + + +def _validate_metadata(metadata_str: str) -> Optional[dict]: + """Validate JSON metadata string. Must be shallow (one level deep max).""" + try: + data = json.loads(metadata_str) + except (json.JSONDecodeError, TypeError): + logger.warning("[artifact_ops] Invalid metadata JSON string") + return None + + if not isinstance(data, dict): + return None + + for value in data.values(): + if isinstance(value, (dict, list)): + return None + + return data + + +def _resolve_branch_name(mention: str) -> Optional[str]: + """Resolve a @mention to a branch name.""" + name = mention.lstrip("@").upper() + + if not os.path.exists(BRANCH_REGISTRY_PATH): + return None + + try: + with open(BRANCH_REGISTRY_PATH, encoding="utf-8") as f: + registry = json.load(f) + for branch in registry.get("branches", []): + if branch.get("name") == name: + return name + return None + except Exception: + logger.error("[artifact_ops] Failed to resolve branch name from registry") + return None + + +# ============================================================================= +# ARTIFACT OPERATIONS +# ============================================================================= + + +def craft_artifact(args: List[str]) -> dict: + """ + Create a new artifact. + + Usage: commons craft "name" "description" [--type crafted] [--rarity common] [--metadata '{}'] + + Returns: + Dict with success, artifact_id, name, type, rarity, creator, description + """ + if not args or len(args) < 2: + return { + "success": False, + "error": 'Usage: commons craft "name" "description" [--type TYPE] [--rarity RARITY]', + } + + name = args[0] + description = args[1] + + artifact_type = "crafted" + rarity = "common" + metadata_str = "{}" + remaining = args[2:] + + i = 0 + while i < len(remaining): + if remaining[i] == "--type" and i + 1 < len(remaining): + artifact_type = remaining[i + 1] + i += 2 + elif remaining[i] == "--rarity" and i + 1 < len(remaining): + rarity = remaining[i + 1] + i += 2 + elif remaining[i] == "--metadata" and i + 1 < len(remaining): + metadata_str = remaining[i + 1] + i += 2 + else: + i += 1 + + if artifact_type not in VALID_TYPES: + return {"success": False, "error": f"Invalid type '{artifact_type}'. Must be one of: {', '.join(VALID_TYPES)}"} + + if rarity not in VALID_RARITIES: + return {"success": False, "error": f"Invalid rarity '{rarity}'. Must be one of: {', '.join(VALID_RARITIES)}"} + + metadata = _validate_metadata(metadata_str) + if metadata is None: + return {"success": False, "error": "Invalid metadata: must be valid shallow JSON (no nested objects/arrays)"} + + from aipass.commons.apps.modules.commons_identity import get_caller_branch + + caller = get_caller_branch() + if not caller: + return {"success": False, "error": "Could not detect calling branch. Run from a branch directory."} + + creator = caller["name"] + + try: + conn = get_db() + + cursor = conn.execute( + "INSERT INTO artifacts (name, type, creator, owner, rarity, description, metadata) " + "VALUES (?, ?, ?, ?, ?, ?, ?)", + (name, artifact_type, creator, creator, rarity, description, json.dumps(metadata)), + ) + artifact_id = cursor.lastrowid + + conn.execute( + "INSERT INTO artifact_history (artifact_id, action, from_agent, to_agent, details) " + "VALUES (?, 'created', ?, ?, ?)", + (artifact_id, creator, creator, f"Crafted '{name}' ({rarity} {artifact_type})"), + ) + + conn.commit() + close_db(conn) + logger.info( + "[commons.artifacts] crafted %r rarity=%s creator=%s", + name, + rarity, + creator, + ) + json_handler.log_operation( + "craft_artifact", + {"artifact_id": artifact_id, "name": name, "rarity": rarity, "creator": creator}, + ) + + return { + "success": True, + "artifact_id": artifact_id, + "name": name, + "type": artifact_type, + "rarity": rarity, + "creator": creator, + "description": description, + } + + except Exception as e: + logger.error(f"Artifact creation failed: {e}") + return {"success": False, "error": str(e)} + + +def list_artifacts(args: List[str]) -> dict: + """ + List artifacts. Default: show only YOUR artifacts. + + Usage: commons artifacts [--all] [--type TYPE] [--rarity RARITY] + + Returns: + Dict with success, artifacts list, scope label, show_all flag + """ + show_all = "--all" in args + filter_type = None + filter_rarity = None + + i = 0 + while i < len(args): + if args[i] == "--type" and i + 1 < len(args): + filter_type = args[i + 1] + i += 2 + elif args[i] == "--rarity" and i + 1 < len(args): + filter_rarity = args[i + 1] + i += 2 + else: + i += 1 + + owner_filter = None + if not show_all: + from aipass.commons.apps.modules.commons_identity import get_caller_branch + + caller = get_caller_branch() + if not caller: + return {"success": False, "error": "Could not detect calling branch. Use --all to see all artifacts."} + owner_filter = caller["name"] + + try: + conn = get_db() + + query = "SELECT id, name, type, creator, owner, rarity, description, created_at FROM artifacts WHERE 1=1" + params: list = [] + + if owner_filter: + query += " AND owner = ?" + params.append(owner_filter) + if filter_type: + query += " AND type = ?" + params.append(filter_type) + if filter_rarity: + query += " AND rarity = ?" + params.append(filter_rarity) + + query += " ORDER BY created_at DESC" + + rows = conn.execute(query, params).fetchall() + close_db(conn) + + except Exception as e: + logger.error(f"Artifact listing failed: {e}") + return {"success": False, "error": str(e)} + + artifacts = [dict(r) for r in rows] + scope_label = "All Artifacts" if show_all else f"Artifacts owned by {owner_filter}" + + return { + "success": True, + "artifacts": artifacts, + "scope_label": scope_label, + "show_all": show_all, + "owner_filter": owner_filter, + } + + +def inspect_artifact(args: List[str]) -> dict: + """ + Show full artifact details including provenance chain. + + Usage: commons inspect [--full] + + Returns: + Dict with success, artifact, history, show_full + """ + if not args: + return {"success": False, "error": "Usage: commons inspect [--full]"} + + show_full = "--full" in args + filtered_args = [a for a in args if a != "--full"] + + if not filtered_args: + return {"success": False, "error": "Usage: commons inspect [--full]"} + + try: + artifact_id = int(filtered_args[0]) + except ValueError: + logger.warning("[artifact_ops] Non-numeric artifact ID provided for inspect") + return {"success": False, "error": "Artifact ID must be a number"} + + try: + conn = get_db() + + row = conn.execute("SELECT * FROM artifacts WHERE id = ?", (artifact_id,)).fetchone() + if not row: + close_db(conn) + return {"success": False, "error": f"Artifact {artifact_id} not found"} + + artifact = dict(row) + + history_rows = conn.execute( + "SELECT * FROM artifact_history WHERE artifact_id = ? ORDER BY created_at ASC", + (artifact_id,), + ).fetchall() + history = [dict(r) for r in history_rows] + + close_db(conn) + + except Exception as e: + logger.error(f"Artifact inspect failed: {e}") + return {"success": False, "error": str(e)} + + # Parse metadata + try: + metadata = json.loads(artifact["metadata"]) if artifact["metadata"] else {} + except (json.JSONDecodeError, TypeError): + logger.warning("[artifact_ops] Failed to parse artifact metadata JSON") + metadata = {} + + artifact["_parsed_metadata"] = metadata + + return { + "success": True, + "artifact": artifact, + "history": history, + "show_full": show_full, + } + + +def collab_artifact(args: List[str]) -> dict: + """ + Initiate a joint artifact that requires multiple signers. + + Usage: commons collab "artifact_name" "description" @signer1 @signer2 [--rarity rare] + + Returns: + Dict with success, pending_id, name, rarity, initiator, signers, expires_at + """ + if len(args) < 3: + return { + "success": False, + "error": 'Usage: commons collab "name" "description" @signer1 @signer2 [--rarity rare]', + } + + artifact_name = args[0] + description = args[1] + + rarity = "rare" + signers = [] + remaining = args[2:] + warnings = [] + i = 0 + while i < len(remaining): + if remaining[i] == "--rarity" and i + 1 < len(remaining): + rarity = remaining[i + 1] + i += 2 + elif remaining[i].startswith("@"): + resolved = _resolve_branch_name(remaining[i]) + if resolved: + signers.append(resolved) + else: + warnings.append(f"Branch '{remaining[i]}' not found, skipping") + i += 1 + else: + i += 1 + + if not signers: + return {"success": False, "error": "At least one @signer is required"} + + if rarity not in VALID_RARITIES: + return {"success": False, "error": f"Invalid rarity '{rarity}'. Must be one of: {', '.join(VALID_RARITIES)}"} + + from aipass.commons.apps.modules.commons_identity import get_caller_branch + + caller = get_caller_branch() + if not caller: + return {"success": False, "error": "Could not detect calling branch. Run from a branch directory."} + + initiator = caller["name"] + signers = list(dict.fromkeys(s for s in signers if s != initiator)) + if not signers: + return {"success": False, "error": "You need at least one other signer (not yourself)"} + + now = datetime.now(timezone.utc) + expires_at = (now + timedelta(hours=48)).strftime("%Y-%m-%dT%H:%M:%SZ") + + try: + conn = get_db() + + cursor = conn.execute( + "INSERT INTO joint_pending (artifact_name, description, rarity, initiator, " + "required_signers, current_signers, expires_at) VALUES (?, ?, ?, ?, ?, '[]', ?)", + (artifact_name, description, rarity, initiator, json.dumps(signers), expires_at), + ) + pending_id = cursor.lastrowid + conn.commit() + close_db(conn) + + return { + "success": True, + "pending_id": pending_id, + "name": artifact_name, + "rarity": rarity, + "initiator": initiator, + "signers": signers, + "expires_at": expires_at, + "warnings": warnings, + } + + except Exception as e: + logger.error(f"Collab artifact failed: {e}") + return {"success": False, "error": str(e)} + + +def sign_artifact(args: List[str]) -> dict: + """ + Sign a pending joint artifact. + + Usage: commons sign + + Returns: + Dict with success, completed (bool), and relevant details + """ + if not args: + return {"success": False, "error": "Usage: commons sign "} + + try: + pending_id = int(args[0]) + except ValueError: + logger.warning("[artifact_ops] Non-numeric pending ID provided for sign") + return {"success": False, "error": "Pending ID must be a number"} + + from aipass.commons.apps.modules.commons_identity import get_caller_branch + + caller = get_caller_branch() + if not caller: + return {"success": False, "error": "Could not detect calling branch. Run from a branch directory."} + + signer = caller["name"] + + try: + conn = get_db() + + row = conn.execute("SELECT * FROM joint_pending WHERE id = ?", (pending_id,)).fetchone() + if not row: + close_db(conn) + return {"success": False, "error": f"Pending joint artifact {pending_id} not found"} + + pending = dict(row) + + now = datetime.now(timezone.utc) + expires_dt = datetime.strptime(pending["expires_at"], "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc) + if now > expires_dt: + conn.execute("DELETE FROM joint_pending WHERE id = ?", (pending_id,)) + conn.commit() + close_db(conn) + return {"success": False, "error": f"Joint artifact {pending_id} has expired"} + + required_signers = json.loads(pending["required_signers"]) + current_signers = json.loads(pending["current_signers"]) + + if signer not in required_signers: + close_db(conn) + return { + "success": False, + "error": f"You are not a required signer. Required: {', '.join(required_signers)}", + } + + if signer in current_signers: + close_db(conn) + return {"success": False, "error": "You have already signed this artifact"} + + current_signers.append(signer) + conn.execute( + "UPDATE joint_pending SET current_signers = ? WHERE id = ?", + (json.dumps(current_signers), pending_id), + ) + + if set(required_signers).issubset(set(current_signers)): + all_participants = [pending["initiator"]] + current_signers + metadata = json.dumps({"signers": all_participants, "joint": True}) + + cursor = conn.execute( + "INSERT INTO artifacts (name, type, creator, owner, rarity, description, metadata) " + "VALUES (?, 'joint', ?, ?, ?, ?, ?)", + ( + pending["artifact_name"], + pending["initiator"], + pending["initiator"], + pending["rarity"], + pending["description"], + metadata, + ), + ) + artifact_id = cursor.lastrowid + + conn.execute( + "INSERT INTO artifact_history (artifact_id, action, from_agent, to_agent, details) " + "VALUES (?, 'created', ?, ?, ?)", + ( + artifact_id, + pending["initiator"], + pending["initiator"], + f"Joint artifact created by {', '.join(all_participants)}", + ), + ) + + conn.execute("DELETE FROM joint_pending WHERE id = ?", (pending_id,)) + conn.commit() + close_db(conn) + + return { + "success": True, + "completed": True, + "artifact_id": artifact_id, + "name": pending["artifact_name"], + "rarity": pending["rarity"], + "participants": all_participants, + "owner": pending["initiator"], + } + else: + conn.commit() + close_db(conn) + + remaining_signers = [s for s in required_signers if s not in current_signers] + return { + "success": True, + "completed": False, + "pending_id": pending_id, + "signer": signer, + "signed": current_signers, + "remaining": remaining_signers, + } + + except Exception as e: + logger.error(f"Sign artifact failed: {e}") + return {"success": False, "error": str(e)} diff --git a/src/aipass/commons/apps/handlers/artifacts/capsule_ops.py b/src/aipass/commons/apps/handlers/artifacts/capsule_ops.py new file mode 100644 index 00000000..dcd20a56 --- /dev/null +++ b/src/aipass/commons/apps/handlers/artifacts/capsule_ops.py @@ -0,0 +1,220 @@ +# =================== AIPass ==================== +# Name: capsule_ops.py +# Description: Time Capsule Operations Handler +# Version: 1.0.0 +# Created: 2026-03-07 +# Modified: 2026-03-07 +# ============================================= + +""" +Time Capsule Operations Handler + +Implementation logic for sealing, listing, and opening time capsules. +Time capsules are sealed messages that can't be opened until a specified date. +Returns dicts for module display layer. +""" + +from typing import List +from datetime import datetime, timezone, timedelta + +from aipass.prax.apps.modules.logger import system_logger as logger + +from aipass.commons.apps.handlers.database.db import get_db, close_db +from aipass.commons.apps.handlers.json import json_handler + + +# ============================================================================= +# SEAL A TIME CAPSULE +# ============================================================================= + +def seal_capsule(args: List[str]) -> dict: + """ + Seal a time capsule that opens after N days. + + Usage: commons capsule "title" "content" + + Returns: + Dict with success, capsule_id, title, creator, days, opens_at + """ + if len(args) < 3: + return {"success": False, "error": 'Usage: commons capsule "title" "content" '} + + title = args[0] + content = args[1] + + try: + days = int(args[2]) + except ValueError: + logger.warning("[capsule_ops] Non-numeric days value provided for seal") + return {"success": False, "error": "Days must be a number"} + + days = max(1, min(365, days)) + + from aipass.commons.apps.modules.commons_identity import get_caller_branch + caller = get_caller_branch() + if not caller: + return {"success": False, "error": "Could not detect calling branch. Run from a branch directory."} + + creator = caller["name"] + + now = datetime.now(timezone.utc) + opens_at = (now + timedelta(days=days)).strftime("%Y-%m-%dT%H:%M:%SZ") + + try: + conn = get_db() + + cursor = conn.execute( + "INSERT INTO time_capsules (creator, title, content, opens_at) " + "VALUES (?, ?, ?, ?)", + (creator, title, content, opens_at), + ) + capsule_id = cursor.lastrowid + conn.commit() + close_db(conn) + json_handler.log_operation("seal_capsule", {"capsule_id": capsule_id, "creator": creator, "days": days}) + + return { + "success": True, + "capsule_id": capsule_id, + "title": title, + "creator": creator, + "days": days, + "opens_at": opens_at, + } + + except Exception as e: + logger.error(f"Seal capsule failed: {e}") + return {"success": False, "error": str(e)} + + +# ============================================================================= +# LIST TIME CAPSULES +# ============================================================================= + +def list_capsules(args: List[str]) -> dict: + """ + List all time capsules with status info. + + Usage: commons capsules + + Returns: + Dict with success, capsules list + """ + try: + conn = get_db() + + rows = conn.execute( + "SELECT * FROM time_capsules ORDER BY opens_at ASC" + ).fetchall() + + close_db(conn) + + except Exception as e: + logger.error(f"List capsules failed: {e}") + return {"success": False, "error": str(e)} + + now = datetime.now(timezone.utc) + capsules = [] + + for row in rows: + capsule = dict(row) + opens_dt = datetime.strptime(capsule["opens_at"], "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc) + + if capsule["opened"]: + capsule["_status"] = "opened" + capsule["_status_text"] = f"Opened by {capsule['opened_by']}" + elif now >= opens_dt: + capsule["_status"] = "ready" + capsule["_status_text"] = "Ready to open!" + else: + delta = opens_dt - now + days_left = delta.days + hours_left = delta.seconds // 3600 + capsule["_status"] = "sealed" + if days_left > 0: + capsule["_status_text"] = f"Sealed ({days_left}d {hours_left}h remaining)" + else: + capsule["_status_text"] = f"Sealed ({hours_left}h remaining)" + + capsules.append(capsule) + + return {"success": True, "capsules": capsules} + + +# ============================================================================= +# OPEN A TIME CAPSULE +# ============================================================================= + +def open_capsule(args: List[str]) -> dict: + """ + Open a time capsule if its opens_at date has passed. + + Usage: commons open + + Returns: + Dict with success, capsule data, opener, already_opened flag + """ + if not args: + return {"success": False, "error": "Usage: commons open "} + + try: + capsule_id = int(args[0]) + except ValueError: + logger.warning("[capsule_ops] Non-numeric capsule ID provided for open") + return {"success": False, "error": "Capsule ID must be a number"} + + from aipass.commons.apps.modules.commons_identity import get_caller_branch + caller = get_caller_branch() + if not caller: + return {"success": False, "error": "Could not detect calling branch. Run from a branch directory."} + + opener = caller["name"] + + try: + conn = get_db() + + row = conn.execute( + "SELECT * FROM time_capsules WHERE id = ?", (capsule_id,) + ).fetchone() + + if not row: + close_db(conn) + return {"success": False, "error": f"Time capsule {capsule_id} not found"} + + capsule = dict(row) + + if capsule["opened"]: + close_db(conn) + return { + "success": True, + "already_opened": True, + "capsule": capsule, + } + + now = datetime.now(timezone.utc) + opens_dt = datetime.strptime(capsule["opens_at"], "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc) + + if now < opens_dt: + delta = opens_dt - now + days_left = delta.days + hours_left = delta.seconds // 3600 + close_db(conn) + return {"success": False, "error": f"This capsule is still sealed. Opens in {days_left}d {hours_left}h."} + + conn.execute( + "UPDATE time_capsules SET opened = 1, opened_by = ? WHERE id = ?", + (opener, capsule_id), + ) + conn.commit() + close_db(conn) + + return { + "success": True, + "already_opened": False, + "capsule": capsule, + "opener": opener, + } + + except Exception as e: + logger.error(f"Open capsule failed: {e}") + return {"success": False, "error": str(e)} diff --git a/src/aipass/commons/apps/handlers/artifacts/trade_ops.py b/src/aipass/commons/apps/handlers/artifacts/trade_ops.py new file mode 100644 index 00000000..e90f06df --- /dev/null +++ b/src/aipass/commons/apps/handlers/artifacts/trade_ops.py @@ -0,0 +1,564 @@ +# =================== AIPass ==================== +# Name: trade_ops.py +# Description: Trading & Ephemeral Item Operations Handler +# Version: 1.0.0 +# Created: 2026-03-07 +# Modified: 2026-03-07 +# ============================================= + +""" +Trading & Ephemeral Item Operations Handler + +Implementation logic for artifact trading, gifting, ephemeral item drops, +item finding, expired item sweeping, and event artifact minting. +Returns dicts for module display layer. +""" + +import json +import os +from datetime import datetime, timezone, timedelta +from typing import List, Optional + +from aipass.prax.apps.modules.logger import system_logger as logger + +from aipass.commons.apps.handlers.database.db import get_db, close_db +from aipass.commons.apps.handlers.json import json_handler + + +# Constants +def _find_registry_path() -> str: + """Walk up from __file__ to find AIPASS_REGISTRY.json in the project root.""" + current = os.path.dirname(os.path.abspath(__file__)) + for _ in range(10): + candidate = os.path.join(current, "AIPASS_REGISTRY.json") + if os.path.exists(candidate): + return candidate + current = os.path.dirname(current) + return os.path.join(os.path.expanduser("~"), "AIPASS_REGISTRY.json") + + +BRANCH_REGISTRY_PATH = _find_registry_path() + +RARITY_COLORS = { + "common": "white", + "uncommon": "green", + "rare": "blue", + "legendary": "yellow", + "unique": "magenta", +} + + +# ============================================================================= +# HELPER FUNCTIONS +# ============================================================================= + + +def _resolve_branch_name(mention: str) -> Optional[str]: + """Resolve a @mention to a branch name.""" + name = mention.lstrip("@").upper() + + if not os.path.exists(BRANCH_REGISTRY_PATH): + return None + + try: + with open(BRANCH_REGISTRY_PATH, encoding="utf-8") as f: + registry = json.load(f) + for branch in registry.get("branches", []): + if branch.get("name") == name: + return name + return None + except Exception: + logger.error("[trade_ops] Failed to resolve branch name from registry") + return None + + +def _now_utc() -> str: + """Return current UTC time as ISO string.""" + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +# ============================================================================= +# SWEEP EXPIRED ITEMS +# ============================================================================= + + +def sweep_expired() -> int: + """ + Sweep-on-access: delete artifacts where expires_at < now. + + Returns: + Number of artifacts swept + """ + try: + conn = get_db() + now = _now_utc() + + expired = conn.execute( + "SELECT id, name, owner FROM artifacts WHERE expires_at IS NOT NULL AND expires_at < ?", + (now,), + ).fetchall() + + if not expired: + close_db(conn) + return 0 + + count = 0 + for row in expired: + conn.execute( + "DELETE FROM artifact_history WHERE artifact_id = ?", + (row["id"],), + ) + conn.execute("DELETE FROM artifacts WHERE id = ?", (row["id"],)) + logger.info(f"[trade_ops] Swept expired item '{row['name']}' (id={row['id']}, owner={row['owner']})") + count += 1 + + conn.commit() + close_db(conn) + return count + + except Exception as e: + logger.error(f"Sweep expired failed: {e}") + return 0 + + +# ============================================================================= +# GIFT ARTIFACT +# ============================================================================= + + +def gift_artifact(args: List[str]) -> dict: + """ + Gift an artifact to another branch. + + Usage: commons gift @branch + + Returns: + Dict with success, artifact info, sender, recipient + """ + if len(args) < 2: + return {"success": False, "error": "Usage: commons gift @branch"} + + try: + artifact_id = int(args[0]) + except ValueError: + logger.warning("[trade_ops] Non-numeric artifact ID provided for gift") + return {"success": False, "error": "Artifact ID must be a number"} + + recipient = _resolve_branch_name(args[1]) + if not recipient: + return {"success": False, "error": f"Branch '{args[1]}' not found in BRANCH_REGISTRY"} + + from aipass.commons.apps.modules.commons_identity import get_caller_branch + + caller = get_caller_branch() + if not caller: + return {"success": False, "error": "Could not detect calling branch."} + + sender = caller["name"] + + if sender == recipient: + return {"success": False, "error": "You cannot gift an artifact to yourself"} + + try: + conn = get_db() + + row = conn.execute("SELECT * FROM artifacts WHERE id = ?", (artifact_id,)).fetchone() + if not row: + close_db(conn) + return {"success": False, "error": f"Artifact {artifact_id} not found"} + + artifact = dict(row) + + if artifact["owner"] != sender: + close_db(conn) + return {"success": False, "error": f"You don't own artifact {artifact_id}. Only the owner can gift it."} + + conn.execute("UPDATE artifacts SET owner = ? WHERE id = ?", (recipient, artifact_id)) + conn.execute( + "INSERT INTO artifact_history (artifact_id, action, from_agent, to_agent, details) " + "VALUES (?, 'gifted', ?, ?, ?)", + (artifact_id, sender, recipient, f"Gifted '{artifact['name']}' from {sender} to {recipient}"), + ) + conn.commit() + close_db(conn) + + json_handler.log_operation( + "gift_artifact", {"artifact_id": artifact_id, "sender": sender, "recipient": recipient} + ) + return { + "success": True, + "artifact_id": artifact_id, + "name": artifact["name"], + "rarity": artifact["rarity"], + "type": artifact["type"], + "sender": sender, + "recipient": recipient, + } + + except Exception as e: + logger.error(f"Gift artifact failed: {e}") + return {"success": False, "error": str(e)} + + +# ============================================================================= +# TRADE ARTIFACTS +# ============================================================================= + + +def trade_artifact(args: List[str]) -> dict: + """ + Trade artifacts between two branches (mutual exchange). + + Usage: commons trade @branch + + Returns: + Dict with success, both artifact details, sender, partner + """ + if len(args) < 3: + return {"success": False, "error": "Usage: commons trade @branch"} + + try: + your_id = int(args[0]) + their_id = int(args[1]) + except ValueError: + logger.warning("[trade_ops] Non-numeric artifact IDs provided for trade") + return {"success": False, "error": "Artifact IDs must be numbers"} + + partner = _resolve_branch_name(args[2]) + if not partner: + return {"success": False, "error": f"Branch '{args[2]}' not found in BRANCH_REGISTRY"} + + from aipass.commons.apps.modules.commons_identity import get_caller_branch + + caller = get_caller_branch() + if not caller: + return {"success": False, "error": "Could not detect calling branch."} + + sender = caller["name"] + + if sender == partner: + return {"success": False, "error": "You cannot trade with yourself"} + + try: + conn = get_db() + + your_row = conn.execute("SELECT * FROM artifacts WHERE id = ?", (your_id,)).fetchone() + their_row = conn.execute("SELECT * FROM artifacts WHERE id = ?", (their_id,)).fetchone() + + if not your_row: + close_db(conn) + return {"success": False, "error": f"Artifact {your_id} not found"} + if not their_row: + close_db(conn) + return {"success": False, "error": f"Artifact {their_id} not found"} + + your_artifact = dict(your_row) + their_artifact = dict(their_row) + + if your_artifact["owner"] != sender: + close_db(conn) + return {"success": False, "error": f"You don't own artifact {your_id}"} + if their_artifact["owner"] != partner: + close_db(conn) + return {"success": False, "error": f"{partner} doesn't own artifact {their_id}"} + + conn.execute("UPDATE artifacts SET owner = ? WHERE id = ?", (partner, your_id)) + conn.execute("UPDATE artifacts SET owner = ? WHERE id = ?", (sender, their_id)) + + conn.execute( + "INSERT INTO artifact_history (artifact_id, action, from_agent, to_agent, details) " + "VALUES (?, 'traded', ?, ?, ?)", + (your_id, sender, partner, f"Traded '{your_artifact['name']}' to {partner} for '{their_artifact['name']}'"), + ) + conn.execute( + "INSERT INTO artifact_history (artifact_id, action, from_agent, to_agent, details) " + "VALUES (?, 'traded', ?, ?, ?)", + (their_id, partner, sender, f"Traded '{their_artifact['name']}' to {sender} for '{your_artifact['name']}'"), + ) + + conn.commit() + close_db(conn) + + return { + "success": True, + "sender": sender, + "partner": partner, + "your_artifact": { + "id": your_id, + "name": your_artifact["name"], + "rarity": your_artifact["rarity"], + }, + "their_artifact": { + "id": their_id, + "name": their_artifact["name"], + "rarity": their_artifact["rarity"], + }, + } + + except Exception as e: + logger.error(f"Trade artifact failed: {e}") + return {"success": False, "error": str(e)} + + +# ============================================================================= +# DROP EPHEMERAL ITEM +# ============================================================================= + + +def drop_item(args: List[str]) -> dict: + """ + Drop an ephemeral item in a room for anyone to find. + + Usage: commons drop "name" "description" [--expires 5] + + Returns: + Dict with success, artifact_id, name, room, expires info + """ + if len(args) < 3: + return {"success": False, "error": 'Usage: commons drop "name" "description" [--expires 5]'} + + name = args[0] + description = args[1] + room = args[2] + + expires_minutes = 5 + remaining = args[3:] + i = 0 + while i < len(remaining): + if remaining[i] == "--expires" and i + 1 < len(remaining): + try: + expires_minutes = int(remaining[i + 1]) + expires_minutes = max(1, min(1440, expires_minutes)) + except ValueError: + logger.warning("[trade_ops] Non-numeric --expires value provided for drop") + return {"success": False, "error": "--expires must be a number (minutes)"} + i += 2 + else: + i += 1 + + from aipass.commons.apps.modules.commons_identity import get_caller_branch + + caller = get_caller_branch() + if not caller: + return {"success": False, "error": "Could not detect calling branch."} + + creator = caller["name"] + + now = datetime.now(timezone.utc) + expires_at = (now + timedelta(minutes=expires_minutes)).strftime("%Y-%m-%dT%H:%M:%SZ") + + try: + conn = get_db() + + room_row = conn.execute("SELECT name FROM rooms WHERE name = ?", (room,)).fetchone() + if not room_row: + close_db(conn) + return {"success": False, "error": f"Room '{room}' does not exist"} + + cursor = conn.execute( + "INSERT INTO artifacts (name, type, creator, owner, rarity, description, room_found, expires_at) " + "VALUES (?, 'found', ?, ?, 'common', ?, ?, ?)", + (name, creator, creator, description, room, expires_at), + ) + artifact_id = cursor.lastrowid + + conn.execute( + "INSERT INTO artifact_history (artifact_id, action, from_agent, to_agent, details) " + "VALUES (?, 'created', ?, NULL, ?)", + (artifact_id, creator, f"Dropped ephemeral item '{name}' in r/{room} (expires in {expires_minutes}m)"), + ) + + conn.commit() + close_db(conn) + + return { + "success": True, + "artifact_id": artifact_id, + "name": name, + "description": description, + "room": room, + "creator": creator, + "expires_minutes": expires_minutes, + "expires_at": expires_at, + } + + except Exception as e: + logger.error(f"Drop item failed: {e}") + return {"success": False, "error": str(e)} + + +# ============================================================================= +# FIND (PICK UP) EPHEMERAL ITEM +# ============================================================================= + + +def find_item(args: List[str]) -> dict: + """ + Pick up an ephemeral item before it expires. + + Usage: commons find + + Returns: + Dict with success, artifact details, finder + """ + if not args: + return {"success": False, "error": "Usage: commons find "} + + try: + artifact_id = int(args[0]) + except ValueError: + logger.warning("[trade_ops] Non-numeric artifact ID provided for find") + return {"success": False, "error": "Artifact ID must be a number"} + + sweep_expired() + + from aipass.commons.apps.modules.commons_identity import get_caller_branch + + caller = get_caller_branch() + if not caller: + return {"success": False, "error": "Could not detect calling branch."} + + finder = caller["name"] + + try: + conn = get_db() + + row = conn.execute("SELECT * FROM artifacts WHERE id = ?", (artifact_id,)).fetchone() + if not row: + close_db(conn) + return {"success": False, "error": f"Artifact {artifact_id} not found (it may have expired)"} + + artifact = dict(row) + + if artifact["type"] != "found": + close_db(conn) + return { + "success": False, + "error": f"Artifact {artifact_id} is not an ephemeral item (type: {artifact['type']})", + } + + if artifact["expires_at"]: + now = datetime.now(timezone.utc) + expires_dt = datetime.strptime(artifact["expires_at"], "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc) + if now > expires_dt: + conn.execute( + "INSERT INTO artifact_history (artifact_id, action, from_agent, to_agent, details) " + "VALUES (?, 'expired', ?, NULL, ?)", + (artifact_id, artifact["owner"], f"Ephemeral item '{artifact['name']}' expired"), + ) + conn.execute("DELETE FROM artifacts WHERE id = ?", (artifact_id,)) + conn.commit() + close_db(conn) + return {"success": False, "error": f"Artifact {artifact_id} has expired and is no longer available"} + + old_owner = artifact["owner"] + + conn.execute( + "UPDATE artifacts SET owner = ?, expires_at = NULL WHERE id = ?", + (finder, artifact_id), + ) + conn.execute( + "INSERT INTO artifact_history (artifact_id, action, from_agent, to_agent, details) " + "VALUES (?, 'found', ?, ?, ?)", + (artifact_id, old_owner, finder, f"Found by {finder} in r/{artifact['room_found'] or 'unknown'}"), + ) + + conn.commit() + close_db(conn) + + return { + "success": True, + "artifact_id": artifact_id, + "name": artifact["name"], + "description": artifact["description"], + "rarity": artifact["rarity"], + "room_found": artifact["room_found"] or "unknown", + "creator": artifact["creator"], + "finder": finder, + } + + except Exception as e: + logger.error(f"Find item failed: {e}") + return {"success": False, "error": str(e)} + + +# ============================================================================= +# MINT EVENT ARTIFACT +# ============================================================================= + + +def mint_event_artifact(args: List[str]) -> dict: + """ + Mint proof-of-attendance artifacts for an event. + + Usage: commons mint "Event Name" @branch1 @branch2 @branch3 + + Returns: + Dict with success, event_name, minted list of (branch, artifact_id) + """ + if len(args) < 2: + return {"success": False, "error": 'Usage: commons mint "Event Name" @branch1 @branch2 ...'} + + event_name = args[0] + mentions = args[1:] + + branches = [] + warnings = [] + for mention in mentions: + branch = _resolve_branch_name(mention) + if branch: + branches.append(branch) + else: + warnings.append(f"Branch '{mention}' not found, skipping") + + if not branches: + return {"success": False, "error": "No valid branches found. Provide at least one @branch."} + + branches = list(dict.fromkeys(branches)) + + try: + conn = get_db() + + conn.execute( + "INSERT OR IGNORE INTO agents (branch_name, display_name, description) VALUES (?, ?, ?)", + ("THE_COMMONS", "The Commons", "The Commons event host"), + ) + + minted = [] + + for branch in branches: + description = f"Proof of attendance: {event_name}" + + cursor = conn.execute( + "INSERT INTO artifacts (name, type, creator, owner, rarity, description, metadata) " + "VALUES (?, 'event', 'THE_COMMONS', ?, 'rare', ?, ?)", + ( + f"{event_name} - Attendee Badge", + branch, + description, + json.dumps({"event": event_name, "attendee": branch}), + ), + ) + artifact_id = cursor.lastrowid + + conn.execute( + "INSERT INTO artifact_history (artifact_id, action, from_agent, to_agent, details) " + "VALUES (?, 'created', 'THE_COMMONS', ?, ?)", + (artifact_id, branch, f"Event badge minted for '{event_name}'"), + ) + + minted.append({"branch": branch, "artifact_id": artifact_id}) + + conn.commit() + close_db(conn) + + return { + "success": True, + "event_name": event_name, + "minted": minted, + "warnings": warnings, + } + + except Exception as e: + logger.error(f"Mint event artifact failed: {e}") + return {"success": False, "error": str(e)}