fix(commons): track artifacts subsystem swallowed by blanket gitignore
The commons craft/trade/capsule subsystem lives at apps/handlers/artifacts/ but the blanket 'artifacts/' ignore (meant for branch-local runtime dirs) silently excluded it from git. The tracked test_artifacts.py imports it, so CI hit ImportError at collection while local passed (files present locally). - Add *.py-scoped negation in .gitignore (keeps logs/ + __pycache__ ignored) - Track artifact_ops.py, trade_ops.py, capsule_ops.py, __init__.py - 19 test_artifacts.py tests pass; imports resolve Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
AIOSAI
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
010cade60f
commit
a231c7d26e
@@ -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/
|
||||
|
||||
|
||||
@@ -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 <id> [--full]
|
||||
|
||||
Returns:
|
||||
Dict with success, artifact, history, show_full
|
||||
"""
|
||||
if not args:
|
||||
return {"success": False, "error": "Usage: commons inspect <artifact_id> [--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 <artifact_id> [--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 <pending_id>
|
||||
|
||||
Returns:
|
||||
Dict with success, completed (bool), and relevant details
|
||||
"""
|
||||
if not args:
|
||||
return {"success": False, "error": "Usage: commons sign <pending_id>"}
|
||||
|
||||
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)}
|
||||
@@ -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" <days>
|
||||
|
||||
Returns:
|
||||
Dict with success, capsule_id, title, creator, days, opens_at
|
||||
"""
|
||||
if len(args) < 3:
|
||||
return {"success": False, "error": 'Usage: commons capsule "title" "content" <days>'}
|
||||
|
||||
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 <capsule_id>
|
||||
|
||||
Returns:
|
||||
Dict with success, capsule data, opener, already_opened flag
|
||||
"""
|
||||
if not args:
|
||||
return {"success": False, "error": "Usage: commons open <capsule_id>"}
|
||||
|
||||
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)}
|
||||
@@ -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 <artifact_id> @branch
|
||||
|
||||
Returns:
|
||||
Dict with success, artifact info, sender, recipient
|
||||
"""
|
||||
if len(args) < 2:
|
||||
return {"success": False, "error": "Usage: commons gift <artifact_id> @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 <your_artifact_id> <their_artifact_id> @branch
|
||||
|
||||
Returns:
|
||||
Dict with success, both artifact details, sender, partner
|
||||
"""
|
||||
if len(args) < 3:
|
||||
return {"success": False, "error": "Usage: commons trade <your_artifact_id> <their_artifact_id> @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" <room> [--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" <room> [--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 <artifact_id>
|
||||
|
||||
Returns:
|
||||
Dict with success, artifact details, finder
|
||||
"""
|
||||
if not args:
|
||||
return {"success": False, "error": "Usage: commons find <artifact_id>"}
|
||||
|
||||
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)}
|
||||
Reference in New Issue
Block a user