Data Integrity, Security, and Compliance

## 🔒 Data Integrity Manifest

**Purpose**: Cryptographic verification of data integrity (ALCOA+ compliance)

{
  "data_integrity_manifest": {
    "generated_at": "2026-01-01T14:23:45.123456",
    "provenance": {
      "session_id": "session_20260101_142000",
      "sha256": "7f83b1657ff1fc53b92dc18148a1d65dfc2d4b1fa3d677284addd200126d9069",
      "activities": 15,
      "entities": 8
    },
    "input_files": {
      "geo_gse109564.h5ad": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
    },
    "system": {
      "lobster_version": "0.3.4",
      "git_commit": "dd2c126f",
      "python_version": "3.13.9",
      "platform": "darwin"
    }
  }
}

"provenance": {
  "session_id": "session_20260101_142000",
  "sha256": "7f83b165...",
  "activities": 15,
  "entities": 8
}

"input_files": {
  "geo_gse109564.h5ad": "e3b0c442...",
  "geo_gse109564_filtered.h5ad": "5d41402a..."
}

"system": {
  "lobster_version": "0.3.4",
  "git_commit": "dd2c126f",
  "python_version": "3.13.9",
  "platform": "darwin"
}

shasum -a 256 geo_gse109564.h5ad
e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855

Get-FileHash -Algorithm SHA256 geo_gse109564.h5ad

import hashlib

def verify_file_hash(filepath, expected_hash):
    """Verify file matches expected SHA-256 hash."""
    sha256 = hashlib.sha256()
    with open(filepath, "rb") as f:
        for chunk in iter(lambda: f.read(4096), b""):
            sha256.update(chunk)

    actual_hash = sha256.hexdigest()
    if actual_hash == expected_hash:
        print("✅ VERIFIED: File hash matches manifest")
        return True
    else:
        print("❌ MISMATCH: File has been modified!")
        print(f"Expected: {expected_hash}")
        print(f"Actual:   {actual_hash}")
        return False

# Usage
verify_file_hash(
    "geo_gse109564.h5ad",
    "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
)

#!/usr/bin/env python3
"""Verify data integrity for Lobster AI notebook."""

import json
import hashlib
import nbformat
from pathlib import Path

def verify_notebook_integrity(notebook_path, data_directory):
    """Verify all input files match manifest hashes."""
    # Read notebook
    with open(notebook_path) as f:
        nb = nbformat.read(f, as_version=4)

    # Find manifest cell
    manifest = None
    for cell in nb.cells:
        if "data_integrity_manifest" in cell.source:
            # Extract JSON from cell
            lines = cell.source.split("\n")
            json_lines = []
            in_json = False
            for line in lines:
                if line.strip().startswith("{"):
                    in_json = True
                if in_json:
                    json_lines.append(line)
                if line.strip().endswith("}") and in_json:
                    break
            manifest = json.loads("\n".join(json_lines))
            break

    if not manifest:
        print("❌ No integrity manifest found in notebook")
        return False

    # Verify each input file
    input_files = manifest["data_integrity_manifest"]["input_files"]
    all_verified = True

    for filename, expected_hash in input_files.items():
        filepath = Path(data_directory) / filename

        if not filepath.exists():
            print(f"⚠️  {filename}: File not found")
            all_verified = False
            continue

        # Calculate hash
        sha256 = hashlib.sha256()
        with open(filepath, "rb") as f:
            for chunk in iter(lambda: f.read(4096), b""):
                sha256.update(chunk)
        actual_hash = sha256.hexdigest()

        if actual_hash == expected_hash:
            print(f"✅ {filename}: Verified")
        else:
            print(f"❌ {filename}: HASH MISMATCH")
            print(f"   Expected: {expected_hash}")
            print(f"   Actual:   {actual_hash}")
            all_verified = False

    return all_verified

# Usage
if __name__ == "__main__":
    import sys
    if len(sys.argv) != 3:
        print("Usage: verify_integrity.py <notebook.ipynb> <data_directory>")
        sys.exit(1)

    verified = verify_notebook_integrity(sys.argv[1], sys.argv[2])
    sys.exit(0 if verified else 1)

python verify_integrity.py my_analysis.ipynb ~/.lobster/

$ shasum -a 256 geo_gse109564.h5ad
e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855

$ shasum -a 256 geo_gse109564.h5ad
a1b2c3d4... (DIFFERENT HASH)

$ shasum -a 256 geo_gse109564.h5ad
shasum: geo_gse109564.h5ad: No such file or directory

# Extract hashes from manifest
# Verify each input file
# Only proceed if all hashes match

analysis_project/
├── my_analysis.ipynb          # Notebook with manifest
├── data/
│   ├── geo_gse109564.h5ad     # Input file
│   └── metadata.csv           # Metadata file
└── verify_integrity.py        # Verification script

Analysis: GSE109564_clustering
Notebook: my_analysis.ipynb
Verified: 2026-01-01 by QA-UserName
Hash Status: ✅ All inputs verified
Reviewer: [Signature]

from lobster.core.utils.h5ad_utils import validate_h5ad

# Validate H5AD file before analysis
is_valid, error_msg = validate_h5ad("geo_gse109564.h5ad")

if is_valid:
    print("✅ H5AD file is valid")
    adata = sc.read_h5ad("geo_gse109564.h5ad")
else:
    print(f"❌ Validation failed: {error_msg}")
    # Handle error (re-download, investigate corruption)

from lobster.core.utils.h5ad_utils import compress_h5ad

# Compress H5AD for archival
original_size = Path("geo_gse109564.h5ad").stat().st_size
compress_h5ad("geo_gse109564.h5ad", compression="gzip", compression_opts=9)
compressed_size = Path("geo_gse109564.h5ad").stat().st_size

print(f"Original: {original_size / 1e9:.2f} GB")
print(f"Compressed: {compressed_size / 1e9:.2f} GB")
print(f"Ratio: {original_size / compressed_size:.1f}x")
# Output: Original: 2.4 GB, Compressed: 0.3 GB, Ratio: 8.0x

def atomic_write_json(path: Path, data: dict):
    """Crash-safe JSON write."""
    temp_path = path.with_suffix('.tmp')

    # Step 1: Write to temp file
    with open(temp_path, 'w') as f:
        json.dump(data, f, indent=2)
        f.flush()
        os.fsync(f.fileno())  # Force write to disk (bypass OS cache)

    # Step 2: Atomic replace (POSIX guarantee)
    os.replace(temp_path, path)  # Atomic on POSIX systems

os.replace(src, dst) on POSIX:
- Atomically replaces dst with src
- If dst exists: overwritten atomically
- If crash occurs: dst is either old OR new (never partial)
- Thread-safe + process-safe (when combined with locks)

# View current session provenance
lobster status

# Export provenance for specific session (W3C-PROV JSON)
# Provenance automatically saved to: .lobster_workspace/provenance.json

[PubMed Search] → [GEO GSE109564 Metadata] → [Download Dataset]
                                               ↓
                                          [geo_gse109564.h5ad]
                                               ↓
                                          [Quality Control]
                                               ↓
                                          [Filter Cells/Genes]
                                               ↓
                                          [Normalization]
                                               ↓
                                          [Clustering]
                                               ↓
                                          [Annotated Dataset]

def analyze(adata: AnnData, **params) -> Tuple[AnnData, Dict[str, Any], AnalysisStep]:
    # ... processing ...
    return processed_adata, stats, analysis_step_ir

AnalysisStep(
    operation="scanpy.tl.leiden",
    tool_name="clustering_service.perform_clustering",
    description="Leiden clustering with resolution 0.5",
    library="scanpy",
    code_template="sc.tl.leiden(adata, resolution={{ resolution }})",
    imports=["import scanpy as sc"],
    parameters={"resolution": 0.5, "random_state": 42},
    parameter_schema={
        "resolution": {"type": "float", "default": 1.0, "min": 0.0},
        "random_state": {"type": "int", "default": 0}
    },
    input_entities=["geo_gse109564_normalized.h5ad"],
    output_entities=["geo_gse109564_clustered.h5ad"]
)

{
  "session_id": "session_20260101_142000",
  "created_at": "2026-01-01T14:20:00.123456Z",
  "last_updated": "2026-01-01T15:45:32.789012Z",
  "workspace_path": "/Users/analyst/.lobster_workspace",
  "modalities": {
    "geo_gse109564": {
      "created_at": "2026-01-01T14:23:15Z",
      "n_obs": 5000,
      "n_vars": 2000,
      "layers": ["counts", "normalized"],
      "last_modified": "2026-01-01T15:30:00Z"
    }
  },
  "tool_usage": [
    {
      "tool_name": "search_pubmed",
      "agent": "research_agent",
      "timestamp": "2026-01-01T14:20:30Z",
      "parameters": {"query": "single-cell CRISPR screening"},
      "result": "Found 15 publications"
    },
    {
      "tool_name": "download_geo_dataset",
      "agent": "data_expert",
      "timestamp": "2026-01-01T14:23:00Z",
      "parameters": {"accession": "GSE109564"},
      "result": "Downloaded 5000 cells × 2000 genes"
    }
  ],
  "agent_handoffs": [
    {
      "from": "supervisor",
      "to": "research_agent",
      "timestamp": "2026-01-01T14:20:15Z",
      "reason": "User requested literature search"
    },
    {
      "from": "research_agent",
      "to": "data_expert",
      "timestamp": "2026-01-01T14:22:45Z",
      "reason": "Download queue entry created for GSE109564"
    }
  ]
}

# Start new session with custom ID
lobster query --session-id "project_gse109564" "Download GSE109564 and cluster"

# Continue previous session
lobster query --session-id latest "Add differential expression analysis"

# View current session status
lobster status

# Export session (includes provenance + metadata)
# Automatically saved to: .lobster_workspace/.session.json

{
  "data_integrity_manifest": {
    "generated_at": "2026-01-01T15:45:00Z",
    "provenance": {
      "session_id": "session_20260101_142000",
      "sha256": "7f83b1657ff1fc53b92dc18148a1d65dfc2d4b1fa3d677284addd200126d9069",
      "activities": 15,
      "entities": 8,
      "agents": 3,
      "time_span": {
        "start": "2026-01-01T14:20:00Z",
        "end": "2026-01-01T15:30:00Z"
      }
    },
    "input_files": { ... },
    "system": { ... }
  }
}

import hashlib
import json

# Read provenance from workspace
with open(".lobster_workspace/provenance.json") as f:
    provenance = json.load(f)

# Compute hash (canonical JSON, sorted keys)
provenance_json = json.dumps(provenance, sort_keys=True)
computed_hash = hashlib.sha256(provenance_json.encode()).hexdigest()

# Compare to manifest hash
manifest_hash = "7f83b1657ff1fc53b92dc18148a1d65dfc2d4b1fa3d677284addd200126d9069"

if computed_hash == manifest_hash:
    print("✅ VERIFIED: Provenance matches notebook manifest")
else:
    print("❌ TAMPERED: Provenance has been modified!")

{
  "cloud_key": "lbstr_abc123...",
  "customer_id": "cust_databiomix",
  "subscription_tier": "premium",
  "features": ["metadata_assistant", "proteomics_expert"],
  "issued_at": "2026-01-01T12:00:00Z",
  "expires_at": "2027-01-01T12:00:00Z",
  "signature": "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9...",
  "revocation_status": {
    "last_checked": "2026-01-01T18:00:00Z",
    "is_revoked": false
  }
}

# Activate cloud license
lobster activate lbstr_abc123...

# Check license status and tier
lobster status

# Output shows:
# Subscription Tier: premium
# Features: metadata_assistant, proteomics_expert
# Expires: 2027-01-01

from lobster.config.subscription_tiers import is_agent_available

# Check if agent is available for current tier
if is_agent_available("metadata_assistant", current_tier="free"):
    # Agent available (False for FREE tier)
    create_agent()
else:
    # Show upgrade message
    print("⚠️ metadata_assistant requires PREMIUM tier")

# supervisor can handoff to research_agent (✅)
# supervisor can handoff to data_expert (✅)
# research_agent CANNOT handoff to metadata_assistant (❌ - PREMIUM only)

User: "Process my publication queue and filter metadata"
Lobster (FREE): "⚠️ Publication queue processing requires PREMIUM tier
                 (metadata_assistant agent). Available with PREMIUM subscription.
                 Visit https://omics-os.com/pricing for upgrade options."

# ✅ GOOD: Use .env file (gitignored)
cat > .env << EOF
ANTHROPIC_API_KEY=sk-ant-api03-...
NCBI_API_KEY=abc123...
EOF

# ✅ GOOD: Workspace-specific keys
mkdir -p ~/project1/.lobster_workspace
echo "ANTHROPIC_API_KEY=sk-ant-project1..." > ~/project1/.env

# ❌ BAD: Hardcode in scripts (committed to git)
export ANTHROPIC_API_KEY="sk-ant-api03-..."  # Will be committed!

# ❌ BAD: Share keys across users
echo "ANTHROPIC_API_KEY=shared-key" > /etc/lobster/.env  # Security risk

# AWS Secrets Manager integration
export ANTHROPIC_API_KEY=$(aws secretsmanager get-secret-value \
  --secret-id lobster/anthropic-key \
  --query SecretString \
  --output text)

# HashiCorp Vault integration
export ANTHROPIC_API_KEY=$(vault kv get -field=api_key secret/lobster/anthropic)

# CI/CD environment injection (GitHub Actions example)
# Stored in repository secrets, injected at runtime
- name: Run Lobster analysis
  env:
    ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
  run: lobster query "Analyze GSE109564"

Development & Exploration → Local Mode
├─ Literature search
├─ Dataset discovery
├─ Small-scale testing
└─ Sensitive data analysis

Production & Scale → Cloud Mode
├─ Large batch processing
├─ Multi-user workflows
├─ Non-sensitive data
└─ Managed infrastructure

# Local mode (default)
unset LOBSTER_CLOUD_KEY
lobster chat

# Cloud mode
export LOBSTER_CLOUD_KEY=lbstr_abc123...
lobster activate $LOBSTER_CLOUD_KEY
lobster chat  # Now uses cloud backend

# Via agent tool
execute_custom_code("""
import pandas as pd

# Load metadata from workspace
metadata = pd.read_csv(WORKSPACE / 'metadata.csv')

# Complex filtering (example: multiple conditions)
filtered = metadata[
    (metadata['disease'] == 'cancer') &
    (metadata['tissue'].isin(['lung', 'breast'])) &
    (metadata['age'] > 50)
]

# Save to exports directory (user-facing files)
OUTPUT_DIR.mkdir(exist_ok=True)
filtered.to_csv(OUTPUT_DIR / 'filtered_metadata.csv', index=False)
""", persist=True)

FORBIDDEN_MODULES = [
    'subprocess',       # Prevent shell command execution
    'os.system',        # Prevent system calls
    'os.popen',         # Prevent pipe-based execution
    'importlib',        # Prevent dynamic imports
    '__import__',       # Prevent import manipulation
    'eval',             # Prevent code evaluation
    'exec',             # Prevent code execution
    'compile',          # Prevent bytecode compilation
    'open',             # Restricted to workspace paths only
]

# ❌ BLOCKED: Attempt to use subprocess
import subprocess
subprocess.run(['rm', '-rf', '/'])  # Detected and blocked at import time

# ❌ BLOCKED: Dynamic import
importlib.import_module('os').system('evil')  # Detected via AST analysis

# ❌ BLOCKED: Code evaluation
eval("__import__('os').system('evil')")  # Detected via AST analysis

# Subprocess killed after timeout
process = subprocess.run(
    [sys.executable, script_path],
    timeout=300,  # 5 minutes default
    capture_output=True,
    text=True,
    cwd=str(workspace_path)  # Workspace boundary
)

# ❌ BLOCKED: Infinite loop (killed after 300s)
while True:
    pass

# ❌ BLOCKED: Excessive computation
for i in range(10**15):
    x = i ** 2  # Killed after timeout

# Custom code executes in workspace directory
# All file paths relative to workspace
WORKSPACE = Path(workspace_path)  # e.g., /Users/analyst/.lobster_workspace
OUTPUT_DIR = WORKSPACE / "exports"  # User-facing files

# Accessing files outside workspace requires absolute paths
# (user's OS permissions apply)

# ✅ ALLOWED: Read/write within workspace
data = pd.read_csv(WORKSPACE / 'metadata.csv')
data.to_csv(OUTPUT_DIR / 'result.csv')

# ⚠️ USER PERMISSION: Access outside workspace (if OS allows)
external_data = pd.read_csv('/external/path/data.csv')  # OS permissions apply

# Every execution logged to provenance
log_tool_usage(
    tool_name="execute_custom_code",
    parameters={"code": code_snippet, "persist": True},
    stats={"execution_time_ms": 1234, "output_lines": 5},
    ir=AnalysisStep(...)  # Complete code captured for reproducibility
)

Academic Lab (trusted users, public data)
→ ✅ Deploy now with Phase 1 security

Biotech Startup (small team, confidential data, on-premise)
→ ✅ Deploy now + code review workflow

Pharma Enterprise (GxP, validated environment)
→ ⚠️ Conditional: Risk assessment + code review + SOP

Cloud SaaS Provider (untrusted users, multi-tenant)
→ ❌ Wait for Phase 2 (Docker isolation)

execute_custom_code("""
import pandas as pd

# Load data
metadata = pd.read_csv(WORKSPACE / 'metadata.csv')

# Filter by condition
cancer_samples = metadata[metadata['disease'] == 'cancer']

# Save to exports (user-facing files)
OUTPUT_DIR.mkdir(exist_ok=True)
cancer_samples.to_csv(OUTPUT_DIR / 'cancer_metadata.csv', index=False)
""", persist=True)

execute_custom_code("""
import pandas as pd
import numpy as np

# Load metadata
meta = pd.read_csv(WORKSPACE / 'metadata.csv')

# Custom QC: Check for missing values
missing_counts = meta.isnull().sum()
qc_pass = missing_counts.max() < 10  # Threshold: < 10 missing per column

# Save QC report
report = pd.DataFrame({
    'column': missing_counts.index,
    'missing_count': missing_counts.values,
    'qc_status': ['PASS' if c < 10 else 'FAIL' for c in missing_counts]
})
report.to_csv(OUTPUT_DIR / 'qc_report.csv', index=False)
""", persist=True)

execute_custom_code("""
import subprocess  # ❌ BLOCKED at import time
subprocess.run(['rm', '-rf', '/'])  # Won't execute

import os
os.system('evil command')  # ❌ BLOCKED (os.system forbidden)

eval("__import__('os').system('evil')")  # ❌ BLOCKED (eval forbidden)
""")

execute_custom_code("""
# ❌ Will timeout after 300 seconds
while True:
    pass

# ❌ Excessive memory usage (may crash subprocess)
data = [0] * (10 ** 10)  # 10 billion elements
""")

# ✅ Verify workspace permissions (user-only access)
chmod 700 ~/.lobster_workspace

# ✅ Set timeout (optional, default 300s)
export LOBSTER_CUSTOM_CODE_TIMEOUT=600  # 10 minutes

# ✅ Enable audit logging (automatically enabled)
lobster query "Your analysis request"

# ✅ Review provenance logs periodically
cat ~/.lobster_workspace/provenance.json | jq '.activities[] | select(.tool_name == "execute_custom_code")'

# ⚠️ Risk assessment (required)
# - Document: Who executes code? What data sensitivity?
# - Review: Security controls sufficient for risk tolerance?

# ✅ Code review workflow (recommended)
# - Require peer review for custom code blocks
# - Document in SOP: "Custom code must be reviewed by lead analyst"

# ✅ Periodic audit (quarterly)
# - Review custom code usage via provenance logs
# - Identify patterns, create standard tools for common operations

# ✅ User training
# - Document allowed patterns (data filtering, QC checks)
# - Document forbidden patterns (network access, subprocess)

# Example: Monitor custom code usage
import json
from pathlib import Path

def audit_custom_code_usage(workspace_path):
    """Generate custom code usage report."""
    provenance_path = Path(workspace_path) / "provenance.json"
    with open(provenance_path) as f:
        prov = json.load(f)

    custom_code_activities = [
        act for act in prov.get('activities', [])
        if act.get('tool_name') == 'execute_custom_code'
    ]

    print(f"Total custom code executions: {len(custom_code_activities)}")
    for act in custom_code_activities:
        print(f"  - {act['timestamp']}: {act['agent']} (session: {act['session_id']})")

# Run monthly
audit_custom_code_usage("~/.lobster_workspace")

## SOP: Custom Code Execution in Lobster AI

**Purpose**: Define approved use of custom code execution feature

**Scope**: All analysts using Lobster AI for bioinformatics analysis

**Approved Use Cases**:
1. Complex metadata filtering (>3 conditions)
2. Custom QC checks not covered by standard tools
3. Format conversions for specialized databases

**Forbidden Operations**:
1. Network requests (data exfiltration risk)
2. File access outside workspace (data leakage risk)
3. Subprocess execution (security risk)

**Review Process**:
1. Analyst documents custom code in lab notebook
2. Lead analyst reviews code before execution
3. QA team audits custom code usage quarterly

**Audit Trail**:
- All custom code logged to provenance.json
- Captured in notebook exports (Data Integrity Manifest)
- Reviewable via `lobster status` command

.lobster_workspace/
├── .session.json              # Current session metadata (multi-process safe)
├── provenance.json            # W3C-PROV audit trail
├── dataset_name.h5ad          # Modality data files
├── plots/                     # Visualizations (PNG, HTML)
│   ├── umap_plot.html
│   └── qc_metrics.png
├── exports/                   # User-facing files (v2.4+)
│   ├── metadata_filtered.csv  # Exported metadata
│   └── de_results.tsv         # Differential expression
├── literature/                # Cached papers (PDF, TXT)
│   └── PMID_12345678.txt
├── metadata/                  # Sample metadata (CSV, TSV)
│   └── GSE109564_metadata.csv
├── download_queue.jsonl       # Download coordination (multi-process safe)
└── publication_queue.jsonl    # Publication processing (multi-process safe)

# ✅ GOOD: Dedicated workspace per project
lobster chat --workspace ~/projects/gse109564_analysis/

# ✅ GOOD: Set global workspace for long session
export LOBSTER_WORKSPACE=~/current_project
lobster chat

# ✅ GOOD: Archive complete analysis
tar -czf gse109564_analysis.tar.gz ~/projects/gse109564_analysis/

# ❌ BAD: Mix multiple projects in same workspace
lobster query "Analyze GSE12345"
lobster query "Analyze GSE67890"  # Mixed in same workspace, confusing provenance

# Each user gets isolated workspace
export LOBSTER_WORKSPACE=/shared/workspaces/$USER
chmod 700 /shared/workspaces/$USER  # User-only access

# Or project-based workspaces
export LOBSTER_WORKSPACE=/shared/projects/project_123
# Set group permissions for team collaboration
chmod 770 /shared/projects/project_123
chgrp bioinfo_team /shared/projects/project_123

# POSIX (macOS, Linux): fcntl.flock
# Windows: msvcrt.locking

class InterProcessFileLock:
    """File-based lock for cross-process coordination."""
    def __enter__(self):
        if platform.system() == 'Windows':
            msvcrt.locking(self.fd, msvcrt.LK_LOCK, 1)
        else:
            fcntl.flock(self.fd, fcntl.LOCK_EX)  # Exclusive lock

from contextlib import contextmanager

@contextmanager
def queue_file_lock(thread_lock, lock_path):
    """Combines threading.Lock + file lock."""
    with thread_lock:  # Thread-safe within process
        with InterProcessFileLock(lock_path):  # Process-safe across processes
            yield

def atomic_write_json(path: Path, data: dict):
    """Crash-safe JSON write (temp file + fsync + atomic replace)."""
    temp_path = path.with_suffix('.tmp')

    # Write to temp file
    with open(temp_path, 'w') as f:
        json.dump(data, f, indent=2)
        f.flush()
        os.fsync(f.fileno())  # Force write to disk

    # Atomic replace (POSIX guarantees atomicity)
    os.replace(temp_path, path)

# Thread A and Thread B both trying to update queue
with queue_file_lock(self._lock, self._lock_path):
    # Critical section - guaranteed exclusive access
    entries = self._read_all_entries()
    entries.append(new_entry)
    atomic_write_jsonl(self._queue_path, entries)
# Lock released - other threads/processes can proceed

Session Creation → Multi-Turn Analysis → Session Snapshot → Restoration
     ↓                    ↓                    ↓                ↓
session_123.json    Updates per tool      .session.json     Resume analysis
  (metadata)         (provenance log)       (checkpoint)     (--session-id)

{
  "session_id": "session_20260101_142000",
  "created_at": "2026-01-01T14:20:00.123456Z",
  "last_updated": "2026-01-01T15:45:32.789012Z",
  "workspace_path": "/Users/analyst/.lobster_workspace",
  "subscription_tier": "premium",
  "modalities": {
    "geo_gse109564": {
      "created_at": "2026-01-01T14:23:15Z",
      "n_obs": 5000,
      "n_vars": 2000,
      "layers": ["counts", "normalized"],
      "last_modified": "2026-01-01T15:30:00Z",
      "file_path": "geo_gse109564.h5ad",
      "size_bytes": 12345678
    }
  },
  "tool_usage_count": 15,
  "agent_handoffs_count": 4
}

# Start new named session
lobster query --session-id "project_gse109564" "Download GSE109564 and cluster"

# Continue most recent session
lobster query --session-id latest "Add differential expression"

# Resume specific session
lobster query --session-id "session_20260101_142000" "Export results"

# View current session
lobster status
# Output:
# Session ID: session_20260101_142000
# Workspace: /Users/analyst/.lobster_workspace
# Modalities: 3 datasets loaded
# Tool usage: 15 operations
# Last updated: 2026-01-01 15:45:32 UTC

# Automatically handled by DataManagerV2
class DataManagerV2:
    def restore_session(self, session_id: str):
        """Restore complete session state."""
        # 1. Load session metadata
        session_meta = self._load_session_file(session_id)

        # 2. Restore modalities (lazy loading)
        for modality_name, info in session_meta['modalities'].items():
            self.modalities[modality_name] = self._load_h5ad(info['file_path'])

        # 3. Restore provenance context
        self.provenance.load_session_activities(session_id)

        # 4. Resume analysis
        return session_meta

# Development: No Redis needed (warning only)
lobster query "Search PubMed for CRISPR papers"
# Warning: Redis not available, rate limiting disabled

# Production: Redis for multi-process coordination
docker run -d -p 6379:6379 redis:alpine
export REDIS_URL=redis://localhost:6379
lobster query "Search PubMed for CRISPR papers"
# Rate limits enforced across all processes

from lobster.tools.rate_limiter import get_rate_limiter

# Decorator automatically applied to provider methods
@get_rate_limiter().with_rate_limit(domain="ncbi")
def search_pubmed(query: str):
    # Rate limit enforced before API call
    # If limit exceeded: waits for token availability
    return ncbi_api.search(query)

rate_limit:ncbi       → Token bucket for NCBI (10 req/s with API key)
rate_limit:pmc        → Token bucket for PMC (3 req/s)
rate_limit:geo        → Token bucket for GEO (10 req/s)
rate_limit:pride      → Token bucket for PRIDE (2 req/s)

# 1. Register at: https://www.ncbi.nlm.nih.gov/account/settings/
# 2. Generate API key
# 3. Add to .env file
echo "NCBI_API_KEY=your-key-here" >> .env

# 4. Verify (rate limit increases automatically)
lobster query "Search PubMed for 500 papers on CRISPR"
# Completes in ~30s instead of ~100s

# Automatic retry on 429 Too Many Requests
@rate_limiter.with_rate_limit(domain="ncbi")
def fetch_data(accession):
    """Automatically retries with exponential backoff."""
    # Retry schedule: 1s, 2s, 4s, 8s, 16s (max 5 attempts)
    # Total wait time: ~31s max
    return api.get(accession)

# Example internal implementation
def _handle_rate_limit_error(response, attempt):
    """Handle 429 Too Many Requests."""
    if response.status_code == 429:
        retry_after = response.headers.get('Retry-After', 2 ** attempt)
        logger.warning(f"Rate limit exceeded, retrying in {retry_after}s")
        time.sleep(retry_after)
        return True  # Retry
    return False  # Don't retry

User: "Search PubMed for CRISPR papers, download top 10 datasets from GEO, fetch protein structures from PDB"
    ↓
research_agent → NCBI rate limiter (10 req/s with key)
    ↓
data_expert → GEO rate limiter (10 req/s)
    ↓
protein_structure_visualization_expert → PDB rate limiter (10 req/s)

Each domain has independent token bucket → No cross-domain interference

# Example: Monitor rate limit usage
def check_rate_limit_health():
    """Check Redis connection and rate limit status."""
    limiter = get_rate_limiter()

    if not limiter.is_available():
        logger.error("⚠️ Redis unavailable - rate limiting disabled")
        return False

    # Check token availability for critical domains
    critical_domains = ["ncbi", "geo", "pride"]
    for domain in critical_domains:
        tokens = limiter.get_available_tokens(domain)
        if tokens < 10:  # Low token threshold
            logger.warning(f"⚠️ Low tokens for {domain}: {tokens}")

    return True

# Run periodically (e.g., every 5 minutes)
check_rate_limit_health()

# Provider-specific timeouts (in provider classes)
class PubMedProvider:
    DEFAULT_TIMEOUT = (10, 30)  # (connect, read) in seconds

    def search(self, query: str):
        try:
            response = requests.get(
                url,
                timeout=self.DEFAULT_TIMEOUT,
                headers={"User-Agent": "Lobster-AI/0.3.4"}
            )
            response.raise_for_status()
            return response.json()
        except requests.Timeout:
            logger.warning("PubMed search timed out, retrying...")
            # Automatic retry via decorator
        except requests.ConnectionError:
            logger.error("Connection to PubMed failed")
            raise ProviderError("Network error")

# Automatic retry logic (internal)
def retry_with_backoff(func, max_attempts=5):
    """Exponential backoff: 1s, 2s, 4s, 8s, 16s."""
    for attempt in range(max_attempts):
        try:
            return func()
        except (requests.Timeout, requests.ConnectionError) as e:
            if attempt == max_attempts - 1:
                raise  # Final attempt failed
            wait_time = 2 ** attempt
            logger.warning(f"Retry {attempt + 1}/{max_attempts} in {wait_time}s")
            time.sleep(wait_time)

User: "Download GSE12345 from GEO"

# Scenario 1: Network timeout
research_agent: "⚠️ Network timeout while accessing GEO. Retrying... (attempt 1/5)"
# ... exponential backoff ...
research_agent: "✅ Successfully downloaded GSE12345 (retry 3 succeeded)"

# Scenario 2: Rate limit exceeded
research_agent: "⚠️ Rate limit exceeded (429). Waiting 8 seconds before retry..."
# ... automatic backoff ...
research_agent: "✅ Request succeeded after rate limit backoff"

# Scenario 3: Permanent failure
research_agent: "❌ Failed to download GSE12345 after 5 attempts. GEO may be unavailable. Please try again later."

# Check recent network errors
cat ~/.lobster_workspace/provenance.json | \
  jq '.activities[] | select(.status == "error") | .error_message'

# Count errors by type
cat ~/.lobster_workspace/provenance.json | \
  jq '.activities[] | select(.status == "error") | .error_type' | \
  sort | uniq -c

from lobster.core.schemas.transcriptomics_schema import TranscriptomicsMetadata

# Validate H5AD metadata before loading
class TranscriptomicsMetadata(BaseModel):
    n_obs: int = Field(gt=0, description="Number of observations (cells)")
    n_vars: int = Field(gt=0, description="Number of variables (genes)")
    layers: List[str] = Field(..., description="Required layers")
    obs_columns: List[str] = Field(..., description="Observation annotations")

    @validator("layers")
    def check_required_layers(cls, v):
        required = ["counts"]
        if not any(layer in v for layer in required):
            raise ValueError(f"Missing required layer: {required}")
        return v

    @validator("n_obs")
    def check_minimum_cells(cls, v):
        if v < 10:
            raise ValueError(f"Too few cells: {v} (minimum: 10)")
        return v

User: "Load my single-cell data"
    ↓
data_expert → load_modality()
    ↓
ModalityAdapter.load() → H5AD file read
    ↓
TranscriptomicsMetadata.validate() → Schema checks
    ├─ ✅ PASS → Data loaded into workspace
    └─ ❌ FAIL → ValidationError with details

Example error:
"ValidationError: n_obs=5 (minimum: 10 cells required)"
"ValidationError: Missing required layer: 'counts'"

# ❌ WITHOUT VALIDATION: Silent failures, corrupted analysis
adata = load_data("bad_file.h5ad")  # Missing 'counts' layer
adata = sc.pp.normalize_total(adata)  # KeyError: 'counts'

# ✅ WITH VALIDATION: Early error detection
adata = load_data("bad_file.h5ad")
# ValidationError: Missing required layer: 'counts'
# Fix data before analysis, avoid wasted time

from lobster.core.identifiers.accession_resolver import get_accession_resolver

resolver = get_accession_resolver()

# Detect database type
db = resolver.detect_database("GSE109564")
# Returns: "geo"

# Validate accession
is_valid = resolver.validate("GSE109564", "geo")
# Returns: True

# Extract accessions from text
accessions = resolver.extract_accessions_by_type("Check GSE109564 and SRP123456")
# Returns: {"geo": ["GSE109564"], "sra": ["SRP123456"]}

# Generate URL
url = resolver.get_url("GSE109564", "geo")
# Returns: "https://www.ncbi.nlm.nih.gov/geo/query/acc.cgi?acc=GSE109564"

User: "Download INVALID12345"
    ↓
research_agent → validate accession
    ↓
AccessionResolver.detect_database("INVALID12345")
    ├─ ✅ Matched pattern → Proceed to download
    └─ ❌ No match → "❌ Invalid accession format: INVALID12345"

User: "Download GSE109564"
    ↓
AccessionResolver.detect_database("GSE109564")  → "geo"
    ↓
AccessionResolver.validate("GSE109564", "geo")  → True
    ↓
Create DownloadQueueEntry(accession="GSE109564", database="geo")
    ↓
data_expert → execute_download_from_queue()

# Check if GEO identifier
if resolver.is_geo_identifier("GSE109564"):
    # Handle GEO-specific logic
    pass

# Check if SRA identifier
if resolver.is_sra_identifier("SRP123456"):
    # Handle SRA-specific logic
    pass

# Check if proteomics identifier
if resolver.is_proteomics_identifier("PXD012345"):
    # Handle PRIDE-specific logic
    pass

# ❌ OLD: Hardcoded regex in every provider (duplicated, error-prone)
class GEOProvider:
    def _validate(self, accession):
        if not re.match(r"^GSE\d+$", accession):  # Duplicated across 10+ files
            raise ValueError("Invalid")

# ✅ NEW: Centralized validation (single source of truth)
class GEOProvider:
    def _validate(self, accession):
        if not get_accession_resolver().validate(accession, "geo"):
            raise ValueError("Invalid")

User: "Download GSE109564"
    ↓
Layer 1: Accession format validation
├─ AccessionResolver.validate("GSE109564", "geo")
└─ ✅ PASS: Valid GEO accession

Layer 2: Database detection
├─ AccessionResolver.detect_database("GSE109564")
└─ ✅ PASS: Detected as "geo"

Layer 3: Metadata fetch
├─ GEOProvider.get_metadata("GSE109564")
├─ Returns: {"n_samples": 5000, "organism": "Homo sapiens", "platform": "GPL24676"}
└─ ✅ PASS: Metadata retrieved

Layer 4: Availability check
├─ requests.head(f"https://ftp.ncbi.nlm.nih.gov/geo/series/GSE109nnn/GSE109564/")
└─ ✅ PASS: Status 200 (dataset exists)

Layer 5: Size estimation
├─ Estimated size: 1.2 GB
└─ ✅ PASS: Below 10 GB threshold (no confirmation needed)

Layer 6: Queue uniqueness
├─ DownloadQueue.check_existing("GSE109564")
└─ ✅ PASS: Not already in queue

Result: Create DownloadQueueEntry(status=PENDING)

# Invalid accession format
User: "Download BADFORMAT123"
→ ❌ Rejected at Layer 1: "Invalid accession format: BADFORMAT123"

# Dataset doesn't exist
User: "Download GSE999999999"
→ ❌ Rejected at Layer 4: "Dataset not found: GSE999999999 (404)"

# Already in queue
User: "Download GSE109564" (twice)
→ ⚠️ Skipped at Layer 6: "GSE109564 already in download queue (status: PENDING)"

User: "Download GSE150614"
    ↓
Layer 5: Size estimation
├─ Estimated size: 15 GB
└─ ⚠️ WARNING: Large dataset detected

research_agent: "⚠️ GSE150614 is large (~15 GB). Download may take 10-30 minutes depending on network speed. Proceed? (yes/no)"

User: "yes"
→ ✅ Proceed to download

User: "no"
→ ❌ Download cancelled

{
  "entry_id": "download_20260101_142000",
  "accession": "GSE109564",
  "database": "geo",
  "status": "PENDING",
  "validation_results": {
    "accession_format": "PASS",
    "database_detection": "PASS (geo)",
    "metadata_fetch": "PASS (5000 samples)",
    "availability_check": "PASS (200 OK)",
    "size_estimation": "PASS (1.2 GB)",
    "queue_uniqueness": "PASS"
  },
  "created_at": "2026-01-01T14:20:00Z"
}

FROM python:3.11-slim

# Security: Non-root user
RUN useradd -m -u 1000 lobster
USER lobster

# Install lobster-ai package
RUN pip install --no-cache-dir lobster-ai

# Workspace mounted at runtime
WORKDIR /workspace

ENTRYPOINT ["lobster"]

# Basic usage
docker run -v $(pwd):/workspace omicsos/lobster:latest query "Analyze GSE109564"

# With API keys
docker run \
  -e ANTHROPIC_API_KEY=$ANTHROPIC_API_KEY \
  -e NCBI_API_KEY=$NCBI_API_KEY \
  -v $(pwd):/workspace \
  omicsos/lobster:latest chat

# With persistent workspace
docker run \
  -v $(pwd)/.lobster_workspace:/workspace \
  omicsos/lobster:latest query "Download GSE109564"

# With Redis for rate limiting
docker network create lobster-net
docker run -d --name redis --network lobster-net redis:alpine
docker run \
  --network lobster-net \
  -e REDIS_URL=redis://redis:6379 \
  -v $(pwd):/workspace \
  omicsos/lobster:latest query "Search PubMed"

version: '3.8'

services:
  redis:
    image: redis:alpine
    restart: unless-stopped
    volumes:
      - redis-data:/data
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 30s

  lobster-cli:
    image: omicsos/lobster:latest
    environment:
      - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
      - REDIS_URL=redis://redis:6379
    volumes:
      - ./workspace:/workspace
    depends_on:
      - redis

volumes:
  redis-data:

s3://lobster-workspaces/
├── user1_project_gse109564/           # Workspace prefix
│   ├── .session.json                   # Session metadata
│   ├── provenance.json                 # W3C-PROV audit trail
│   ├── geo_gse109564.h5ad              # Modality data
│   ├── plots/
│   │   └── umap_plot.html
│   └── exports/
│       └── metadata_filtered.csv
├── user2_project_xyz/
│   └── ...

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "s3:GetObject",
        "s3:PutObject",
        "s3:DeleteObject",
        "s3:ListBucket"
      ],
      "Resource": [
        "arn:aws:s3:::lobster-workspaces/user1_*/*",
        "arn:aws:s3:::lobster-workspaces"
      ],
      "Condition": {
        "StringLike": {
          "s3:prefix": ["user1_*"]
        }
      }
    }
  ]
}

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "DenyUnencryptedObjectUploads",
      "Effect": "Deny",
      "Principal": "*",
      "Action": "s3:PutObject",
      "Resource": "arn:aws:s3:::lobster-workspaces/*",
      "Condition": {
        "StringNotEquals": {
          "s3:x-amz-server-side-encryption": "AES256"
        }
      }
    }
  ]
}

# Configure S3 backend (environment variables)
export AWS_ACCESS_KEY_ID=your-access-key
export AWS_SECRET_ACCESS_KEY=your-secret-key
export LOBSTER_S3_BUCKET=lobster-workspaces
export LOBSTER_S3_WORKSPACE_PREFIX=user1_project_gse109564

# Lobster automatically uses S3 backend
lobster query "Analyze GSE109564"
# Data stored to: s3://lobster-workspaces/user1_project_gse109564/

{
  "Rules": [
    {
      "Id": "TransitionToIA",
      "Status": "Enabled",
      "Transitions": [
        {
          "Days": 30,
          "StorageClass": "STANDARD_IA"
        }
      ],
      "NoncurrentVersionTransitions": [
        {
          "NoncurrentDays": 7,
          "StorageClass": "GLACIER"
        }
      ]
    },
    {
      "Id": "ExpireOldVersions",
      "Status": "Enabled",
      "NoncurrentVersionExpiration": {
        "NoncurrentDays": 90
      }
    }
  ]
}