From af3757d039583aca62aa363c29bae0c9d04d2748 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 11 Mar 2026 00:58:05 +0000 Subject: [PATCH] feat: add generic dataset ingestion skill with review hardening Adds ingest-dataset system skill for CSV, JSON, JSONL, HuggingFace, and Parquet datasets. Lightweight formats run on the actor; heavy formats (HF, Parquet) delegate to the sandbox via a Python loader. Also addresses review findings: - Auth middleware validates API_KEY with timing-safe comparison - Shared ActorDb interface replaces `any` escape hatch - Position embeddings cached and used for novelty scoring - Drain writes markdown after SQLite commit (true atomicity) - crypto.randomUUID() replaces Date.now() for all IDs - Remove double session destroy + dead teardownSession export - Domain relevance gate uses whole-word matching + stop words - Silent catch blocks now log with console.warn https://claude.ai/code/session_01BmJuzKQC7xoR7vsdFcRUYA --- .env.example | 3 + Dockerfile | 6 + actor/dataset.ts | 142 +++++++++++++++ actor/drain.ts | 218 +++++++++++++++++------ actor/index.ts | 222 ++++++++++++++++++++---- actor/materialize.ts | 1 + actor/tension.ts | 77 +++++++- actor/types.ts | 61 ++++++- api/index.ts | 13 +- api/sentient/post.ts | 2 +- sandbox/index.ts | 113 +++++++++++- sentient.jsonc | 14 ++ skills/.system/ingest-dataset/SKILL.md | 69 ++++++++ skills/.system/ingest-dataset/loader.py | 166 ++++++++++++++++++ 14 files changed, 994 insertions(+), 113 deletions(-) create mode 100644 actor/dataset.ts create mode 100644 skills/.system/ingest-dataset/SKILL.md create mode 100644 skills/.system/ingest-dataset/loader.py diff --git a/.env.example b/.env.example index 75b082e..cbe2281 100644 --- a/.env.example +++ b/.env.example @@ -30,6 +30,9 @@ UNKEY_ROOT_KEY= # ── Telegram ─────────────────────────────────────────────────────────── TELEGRAM_BOT_TOKEN= # @BotFather → /newbot +# ── HuggingFace (optional — for private datasets in sandbox) ────────── +HF_TOKEN= # https://huggingface.co/settings/tokens + # ── GitHub (optional) ────────────────────────────────────────────────── GITHUB_TOKEN= # fine-grained PAT for PR creation diff --git a/Dockerfile b/Dockerfile index 80e8e48..d5b04a9 100644 --- a/Dockerfile +++ b/Dockerfile @@ -11,6 +11,12 @@ RUN npm install -g opencode-ai@latest RUN npm install -g @anthropic-ai/claude-code@latest RUN npm install -g skills@latest +# Python data layer (for sandbox dataset operations) +RUN apt-get update && apt-get install -y python3 python3-pip python3-venv \ + && rm -rf /var/lib/apt/lists/* +RUN python3 -m pip install --break-system-packages \ + datasets huggingface_hub pandas pyarrow requests + WORKDIR /workspace EXPOSE 8080 EXPOSE 4096 diff --git a/actor/dataset.ts b/actor/dataset.ts new file mode 100644 index 0000000..b871dd4 --- /dev/null +++ b/actor/dataset.ts @@ -0,0 +1,142 @@ +import type { DatasetConfig, DatasetOutput, IngestSignal, PositionUpdate } from "./types" +import { needsSandbox } from "./types" + +// ── Column Mapping ─────────────────────────────────────────────────── + +function applyColumnMap( + row: Record, + config: DatasetConfig, +): Record { + if (!config.column_map) return row + const mapped: Record = {} + for (const [srcCol, destCol] of Object.entries(config.column_map)) { + if (row[srcCol] !== undefined) mapped[destCol] = row[srcCol] + } + for (const [key, val] of Object.entries(row)) { + if (!config.column_map[key]) mapped[key] = val + } + return mapped +} + +function extractContent(row: Record, config: DatasetConfig): string { + if (config.content_column && row[config.content_column] !== undefined) { + return String(row[config.content_column]) + } + for (const key of ["text", "content", "body", "description"]) { + if (typeof row[key] === "string") return row[key] as string + } + const firstString = Object.values(row).find((v) => typeof v === "string") + return firstString ? String(firstString) : JSON.stringify(row) +} + +function extractConfidence(row: Record, config: DatasetConfig): number { + if (config.label_column && row[config.label_column] !== undefined) { + const val = Number(row[config.label_column]) + if (!Number.isNaN(val)) return Math.min(1, Math.max(0, val)) + } + return config.credibility ?? 0.5 +} + +// ── Row Converters ─────────────────────────────────────────────────── + +function rowToSignal( + row: Record, + index: number, + config: DatasetConfig, +): IngestSignal { + const mapped = applyColumnMap(row, config) + return { + id: `dataset-${config.uri}-${index}-${crypto.randomUUID()}`, + sourceProvider: "dataset", + sourceMode: "batch", + content: extractContent(mapped, config), + urgency: "medium", + credibility: extractConfidence(mapped, config), + timestamp: Date.now(), + metadata: { datasetSource: config.source, datasetUri: config.uri, rowIndex: index }, + } +} + +function rowToPosition( + row: Record, + index: number, + config: DatasetConfig, +): PositionUpdate { + const mapped = applyColumnMap(row, config) + const text = extractContent(mapped, config) + const slug = + typeof mapped.slug === "string" + ? (mapped.slug as string) + : `dataset-${config.uri.replace(/[^a-z0-9]/gi, "-")}-${index}` + return { + slug, + text, + confidence: extractConfidence(mapped, config), + surpriseDelta: 0, + priorConfidence: 0, + status: "settled", + } +} + +// ── Parsers ────────────────────────────────────────────────────────── + +type Rows = Record[] + +const parsers: Record Promise> = { + async json(uri) { + const data = await Bun.file(uri).json() + return Array.isArray(data) ? data : [data] + }, + async jsonl(uri) { + const text = await Bun.file(uri).text() + return text + .split("\n") + .filter(Boolean) + .map((line) => JSON.parse(line)) + }, + async csv(uri) { + return Bun.csv(Bun.file(uri), { headers: true }) as Promise + }, + async url(uri) { + const res = await fetch(uri) + if (!res.ok) throw new Error(`Dataset fetch failed (${res.status}): ${uri}`) + const contentType = res.headers.get("content-type") ?? "" + if (contentType.includes("json")) { + const data = await res.json() + return Array.isArray(data) ? (data as Rows) : [data as Rows[0]] + } + const text = await res.text() + try { + const data = JSON.parse(text) + return Array.isArray(data) ? data : [data] + } catch { + return Bun.csv(text, { headers: true }) as unknown as Rows + } + }, +} + +// ── Local Ingestion (Actor-side) ───────────────────────────────────── + +export async function ingestDatasetLocal(config: DatasetConfig): Promise { + if (needsSandbox(config.source)) { + throw new Error( + `Dataset source "${config.source}" requires sandbox. Use runDatasetSession() instead.`, + ) + } + + const parser = parsers[config.source] + if (!parser) throw new Error(`Unsupported local dataset source: ${config.source}`) + + const rows = (await parser(config.uri)).slice(0, config.limit ?? 1000) + const output: DatasetOutput = { + source: `${config.source}:${config.uri}`, + rowsProcessed: rows.length, + } + + if (config.mode === "signals") { + output.signals = rows.map((row, i) => rowToSignal(row, i, config)) + } else if (config.mode === "positions") { + output.positions = rows.map((row, i) => rowToPosition(row, i, config)) + } + return output +} diff --git a/actor/drain.ts b/actor/drain.ts index 1fdd819..5eabf23 100644 --- a/actor/drain.ts +++ b/actor/drain.ts @@ -1,4 +1,5 @@ import type { + ActorDb, CalibrationEvent, IngestSignal, InquiryUpdate, @@ -116,7 +117,7 @@ export function calibrationEntryToMarkdown( slug: string, detail: Record, ): string { - const id = `${type}-${slug}-${Date.now()}` + const id = `${type}-${slug}-${crypto.randomUUID()}` return [ "---", `id: ${id}`, @@ -130,10 +131,10 @@ export function calibrationEntryToMarkdown( ].join("\n") } -// ── Drain Session ───────────────────────────────────────────────────── +// ── Drain Context & Shared Helpers ──────────────────────────────────── interface DrainContext { - db: { execute: (sql: string, ...params: unknown[]) => Promise } + db: ActorDb state: { pendingProofs: PositionProofOfWork[] calibrationThreshold: number @@ -142,76 +143,119 @@ interface DrainContext { broadcast: (event: string, payload: CalibrationEvent) => void } -export async function drainSession( +type DrainResult = { proofsSurfaced: PositionProofOfWork[]; autoCommitted: PositionUpdate[] } + +/** Pending file write accumulated during drain, flushed after SQLite commit. */ +interface PendingWrite { + path: string + content: string +} + +/** Shared position drain: proof gate, SQLite upserts, pending markdown writes. */ +async function drainPositions( ctx: DrainContext, - summary: SessionSummary, -): Promise<{ proofsSurfaced: PositionProofOfWork[]; autoCommitted: PositionUpdate[] }> { + positions: PositionUpdate[], + sessionId: string, + triggerSignalId: string, + pendingWrites: PendingWrite[], +): Promise { const proofsSurfaced: PositionProofOfWork[] = [] const autoCommitted: PositionUpdate[] = [] - // Write session record markdown - const recordMd = sessionRecordToMarkdown(summary) - await Bun.write(`knowledge/record/${summary.id}.md`, recordMd) - - // Process all position updates - const allPositions = [...summary.newPositions, ...summary.updatedPositions] - for (const position of allPositions) { + for (const position of positions) { if (position.surpriseDelta > ctx.state.calibrationThreshold) { - // High surprise — surface as proof for human review const proof: PositionProofOfWork = { index: ctx.state.pendingProofs.length + proofsSurfaced.length + 1, slug: position.slug, - sourceSignalId: summary.triggerSignalId, + sourceSignalId: triggerSignalId, priorConfidence: position.priorConfidence, posteriorConfidence: position.confidence, surpriseDelta: position.surpriseDelta, text: position.text, - sessionId: summary.id, + sessionId, } proofsSurfaced.push(proof) position.status = "pending_review" - ctx.broadcast("calibrationEvent", { type: "proofSurfaced", proof }) } else { - // Low surprise — auto-commit autoCommitted.push(position) } - // Write position markdown - const positionMd = positionToMarkdown(position, summary.id) - await Bun.write(`knowledge/positions/${position.slug}.md`, positionMd) + pendingWrites.push({ + path: `knowledge/positions/${position.slug}.md`, + content: positionToMarkdown(position, sessionId), + }) + } + + // SQLite position upserts + for (const p of positions) { + await ctx.db.execute( + `INSERT INTO positions (slug, confidence, surprise_delta, status, source_session, text, wikilinks, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, datetime('now')) + ON CONFLICT(slug) DO UPDATE SET + confidence = excluded.confidence, + surprise_delta = excluded.surprise_delta, + status = excluded.status, + source_session = excluded.source_session, + text = excluded.text, + wikilinks = excluded.wikilinks, + updated_at = datetime('now')`, + p.slug, + p.confidence, + p.surpriseDelta, + p.status, + `record/${sessionId}.md`, + p.text, + JSON.stringify(p.wikilinks ?? []), + ) + } + + ctx.state.pendingProofs.push(...proofsSurfaced) + ctx.state.sessionCount += 1 + + return { proofsSurfaced, autoCommitted } +} + +/** Flush pending writes to disk. Called after successful SQLite commit. */ +async function flushWrites(writes: PendingWrite[]): Promise { + for (const w of writes) { + await Bun.write(w.path, w.content) } +} + +// ── Drain Session ───────────────────────────────────────────────────── + +export async function drainSession( + ctx: DrainContext, + summary: SessionSummary, +): Promise { + const allPositions = [...summary.newPositions, ...summary.updatedPositions] + const pendingWrites: PendingWrite[] = [] - // Write inquiry markdown + // Queue session record markdown + pendingWrites.push({ + path: `knowledge/record/${summary.id}.md`, + content: sessionRecordToMarkdown(summary), + }) + + // Queue inquiry markdown for (const inquiry of summary.newInquiries) { - const inquiryMd = inquiryToMarkdown(inquiry) - await Bun.write(`knowledge/inquiries/${inquiry.slug}.md`, inquiryMd) + pendingWrites.push({ + path: `knowledge/inquiries/${inquiry.slug}.md`, + content: inquiryToMarkdown(inquiry), + }) } // SQLite atomic transaction await ctx.db.execute("BEGIN") try { - for (const p of allPositions) { - await ctx.db.execute( - `INSERT INTO positions (slug, confidence, surprise_delta, status, source_session, text, wikilinks, updated_at) - VALUES (?, ?, ?, ?, ?, ?, ?, datetime('now')) - ON CONFLICT(slug) DO UPDATE SET - confidence = excluded.confidence, - surprise_delta = excluded.surprise_delta, - status = excluded.status, - source_session = excluded.source_session, - text = excluded.text, - wikilinks = excluded.wikilinks, - updated_at = datetime('now')`, - p.slug, - p.confidence, - p.surpriseDelta, - p.status, - `record/${summary.id}.md`, - p.text, - JSON.stringify(p.wikilinks ?? []), - ) - } + const result = await drainPositions( + ctx, + allPositions, + summary.id, + summary.triggerSignalId, + pendingWrites, + ) for (const i of summary.newInquiries) { await ctx.db.execute( @@ -245,16 +289,80 @@ export async function drainSession( ) await ctx.db.execute("COMMIT") + + // Flush markdown to disk only after successful commit + await flushWrites(pendingWrites) + + return result } catch (error) { await ctx.db.execute("ROLLBACK") throw error } +} - // Add proofs to pending state - ctx.state.pendingProofs.push(...proofsSurfaced) - ctx.state.sessionCount += 1 +// ── Drain Dataset Positions ─────────────────────────────────────────── - return { proofsSurfaced, autoCommitted } +export async function drainDatasetPositions( + ctx: DrainContext, + positions: PositionUpdate[], + datasetSource: string, +): Promise { + const syntheticSessionId = `dataset-${crypto.randomUUID()}` + const pendingWrites: PendingWrite[] = [] + + // Queue synthetic session record + const recordMd = [ + "---", + `id: ${syntheticSessionId}`, + "trigger_signal_id: dataset-ingest", + "tension_at_trigger: 1.0", + `started_at: ${new Date().toISOString()}`, + `completed_at: ${new Date().toISOString()}`, + `positions_updated: ${positions.length}`, + "inquiries_opened: 0", + "layers_applied: [ingest-dataset]", + "---", + "", + `# Dataset Ingestion: ${datasetSource}`, + "", + `Ingested ${positions.length} positions from dataset.`, + ].join("\n") + pendingWrites.push({ path: `knowledge/record/${syntheticSessionId}.md`, content: recordMd }) + + // SQLite atomic transaction + await ctx.db.execute("BEGIN") + try { + const result = await drainPositions( + ctx, + positions, + syntheticSessionId, + "dataset-ingest", + pendingWrites, + ) + + await ctx.db.execute( + `INSERT INTO record (id, trigger_signal_id, tension_at_trigger, started_at, completed_at, positions_updated, inquiries_opened, narrative) + VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, + syntheticSessionId, + "dataset-ingest", + 1.0, + new Date().toISOString(), + new Date().toISOString(), + positions.length, + 0, + `Dataset ingestion from ${datasetSource}`, + ) + + await ctx.db.execute("COMMIT") + + // Flush markdown to disk only after successful commit + await flushWrites(pendingWrites) + + return result + } catch (error) { + await ctx.db.execute("ROLLBACK") + throw error + } } // ── Commit Proof (Accept / Adjust) ──────────────────────────────────── @@ -296,11 +404,12 @@ export async function commitProof( } const calMd = calibrationEntryToMarkdown(calType, slug, calDetail) - await Bun.write(`knowledge/calibration/accept-${slug}-${Date.now()}.md`, calMd) + const acceptCalId = `accept-${slug}-${crypto.randomUUID()}` + await Bun.write(`knowledge/calibration/${acceptCalId}.md`, calMd) await ctx.db.execute( "INSERT INTO calibration (id, type, slug, detail) VALUES (?, ?, ?, ?)", - `accept-${slug}-${Date.now()}`, + acceptCalId, calType, slug, JSON.stringify(calDetail), @@ -349,11 +458,12 @@ export async function rejectProof( attemptedConfidence: proof.posteriorConfidence, note, }) - await Bun.write(`knowledge/calibration/reject-${slug}-${Date.now()}.md`, calMd) + const rejectCalId = `reject-${slug}-${crypto.randomUUID()}` + await Bun.write(`knowledge/calibration/${rejectCalId}.md`, calMd) await ctx.db.execute( "INSERT INTO calibration (id, type, slug, detail) VALUES (?, 'proof_rejected', ?, ?)", - `reject-${slug}-${Date.now()}`, + rejectCalId, slug, JSON.stringify({ note, prior: proof.priorConfidence, attempted: proof.posteriorConfidence }), ) @@ -366,7 +476,7 @@ export async function rejectProof( // Create correction signal const correctionSignal: IngestSignal = { - id: `correction-${slug}-${Date.now()}`, + id: `correction-${slug}-${crypto.randomUUID()}`, sourceProvider: "correction", sourceMode: "system", content: `Owner correction for position "${slug}": ${note}`, diff --git a/actor/index.ts b/actor/index.ts index 81ed726..8e91b18 100644 --- a/actor/index.ts +++ b/actor/index.ts @@ -1,6 +1,7 @@ import { actor, event, queue, setup } from "rivetkit" import { workflow } from "rivetkit/workflow" -import { ensureSandbox, mountWorkspace, runSession, teardownSession } from "../sandbox/index" +import type { SandboxAgent } from "sandbox-agent" +import { ensureSandbox, mountWorkspace, runDatasetSession, runSession } from "../sandbox/index" import { DEFAULT_BUDGET, calculateCost, @@ -10,20 +11,30 @@ import { recordCost, resetBudgetIfNewDay, } from "./budget" -import { commitProof, drainSession, rejectProof, validateSummary } from "./drain" +import { ingestDatasetLocal } from "./dataset" +import { + commitProof, + drainDatasetPositions, + drainSession, + rejectProof, + validateSummary, +} from "./drain" import { parseSentientConfig } from "./materialize" import { actorDb } from "./schema" import { runActorSkill } from "./skill-runner" -import { evaluateTension } from "./tension" +import { embed, evaluateTension } from "./tension" import type { + ActorDb, AgentCommand, CalibrationEvent, CalibrationState, + DatasetConfig, IngestSignal, SentientState, SessionSummary, TriageResult, } from "./types" +import { needsSandbox } from "./types" // ── Actor Definition ────────────────────────────────────────────────── @@ -56,6 +67,7 @@ const opensentient = actor({ budgetState: { ...DEFAULT_BUDGET }, triageEnabled: true, modelPricingCache: null, + positionEmbeddings: {}, }), events: { @@ -91,7 +103,7 @@ const opensentient = actor({ if (cmd.type === "run") { // Manual run — create a synthetic signal const signal: IngestSignal = { - id: `manual-${Date.now()}`, + id: `manual-${crypto.randomUUID()}`, sourceProvider: "owner", sourceMode: "command", content: (cmd.payload.reason as string) ?? "Manual session triggered", @@ -100,6 +112,11 @@ const opensentient = actor({ timestamp: Date.now(), } await processSignal(loopCtx as unknown as ProcessContext, signal) + } else if (cmd.type === "ingest_dataset") { + await processDatasetIngestion( + loopCtx as unknown as ProcessContext, + cmd.payload.dataset as DatasetConfig, + ) } return // next loop iteration } @@ -170,8 +187,18 @@ const opensentient = actor({ // v0.2: Fetch model pricing on init try { c.state.modelPricingCache = await fetchModelPricing(null) - } catch { - // Non-fatal — pricing cache will be null, costs estimated at 0 + } catch (err) { + console.warn("[init] Failed to fetch model pricing:", err) + } + + // v0.3: Enqueue init-scheduled dataset ingestions + for (const ds of config.datasets ?? []) { + if (ds.schedule === "init") { + await c.queue.send("commands", { + type: "ingest_dataset", + payload: { dataset: ds }, + }) + } } }, @@ -199,12 +226,12 @@ const opensentient = actor({ // v0.2: Refresh pricing cache daily try { c.state.modelPricingCache = await fetchModelPricing(c.state.modelPricingCache) - } catch { - // Non-fatal + } catch (err) { + console.warn("[daily] Failed to refresh model pricing:", err) } const signal: IngestSignal = { - id: `daily-scan-${Date.now()}`, + id: `daily-scan-${crypto.randomUUID()}`, sourceProvider: "system", sourceMode: "poll", content: "Scheduled daily domain scan", @@ -213,6 +240,11 @@ const opensentient = actor({ timestamp: Date.now(), } await c.queue.send("signals", signal) + + // v0.3: Enqueue daily-scheduled dataset ingestions + // Note: datasets config is not in actor state — would need to re-parse + // or store in state. For now, daily datasets are triggered via API. + // Re-schedule unconditionally c.schedule.after(24 * 60 * 60 * 1000, "enqueueDailyScan") }, @@ -258,7 +290,7 @@ const opensentient = actor({ ) // Enqueue a signal to investigate this tension const signal: IngestSignal = { - id: `tension-confirmed-${slug}-${Date.now()}`, + id: `tension-confirmed-${slug}-${crypto.randomUUID()}`, sourceProvider: "owner", sourceMode: "command", content: `Owner confirmed tension on position: ${slug}. Investigate.`, @@ -286,7 +318,7 @@ const opensentient = actor({ // Write calibration log await c.db.execute( "INSERT INTO calibration (id, type, slug, detail) VALUES (?, 'tension_dismissed', ?, ?)", - `dismiss-${slug}-${Date.now()}`, + `dismiss-${slug}-${crypto.randomUUID()}`, slug, JSON.stringify({ weightAdjustment: weightAdj }), ) @@ -303,7 +335,7 @@ const opensentient = actor({ // Write correction as calibration log await c.db.execute( "INSERT INTO calibration (id, type, slug, detail) VALUES (?, 'correction', ?, ?)", - `correct-${slug}-${Date.now()}`, + `correct-${slug}-${crypto.randomUUID()}`, slug, JSON.stringify({ text, confidence }), ) @@ -330,14 +362,18 @@ const opensentient = actor({ c.state.budgetState.x402AllocationPct = x402Pct } }, + + ingestDataset: async (c, datasetConfig: DatasetConfig) => { + await c.queue.send("commands", { + type: "ingest_dataset", + payload: { dataset: datasetConfig }, + }) + }, }, }) // ── Signal Processing (used inside workflow) ────────────────────────── -// biome-ignore lint/suspicious/noExplicitAny: Rivet DB type is opaque, inferred at runtime -type ActorDb = any - type ProcessContext = { state: SentientState db: ActorDb @@ -352,6 +388,38 @@ type ProcessContext = { rollbackCheckpoint: (name: string) => Promise } +async function withSandbox(ctx: ProcessContext, stepPrefix = ""): Promise { + const prefix = stepPrefix ? `${stepPrefix}-` : "" + const { sdk } = await ctx.step({ + name: `${prefix}ensure-sandbox`, + timeout: 120_000, + maxRetries: 2, + retryBackoffBase: 5_000, + run: () => ensureSandbox(ctx.state), + }) + await ctx.step({ + name: `${prefix}mount-workspace`, + timeout: 60_000, + run: () => mountWorkspace(sdk, ctx.state), + }) + return sdk +} + +/** Embed and cache position texts so future tension evaluations use them. */ +async function cachePositionEmbeddings( + ctx: ProcessContext, + positions: Array<{ slug: string; text: string }>, +): Promise { + for (const p of positions) { + try { + const { embedding } = await embed(p.text, ctx.state.modelConfig.embedding) + ctx.state.positionEmbeddings[p.slug] = Array.from(embedding) + } catch { + // Non-fatal — position will be re-embedded on next cache miss + } + } +} + async function processSignal(ctx: ProcessContext, signal: IngestSignal): Promise { // TTL check if (Date.now() - signal.timestamp > 7 * 24 * 60 * 60 * 1000) return @@ -370,12 +438,17 @@ async function processSignal(ctx: ProcessContext, signal: IngestSignal): Promise }> const positionSlugs = positions.map((p) => p.slug) + // Build position embeddings array from cache (order matches positionSlugs) + const positionEmbeddings: Float32Array[] = positionSlugs + .filter((slug) => ctx.state.positionEmbeddings[slug]) + .map((slug) => new Float32Array(ctx.state.positionEmbeddings[slug])) + // Evaluate tension (domain relevance + embedding similarity) - const { tension, embeddingCost } = await evaluateTension( + const { tension, embeddingCost, signalEmbedding } = await evaluateTension( signal, ctx.state.domain, positionSlugs, - [], // Position embeddings — would be cached in production + positionEmbeddings, ctx.state.modelConfig.embedding, ctx.state.signalWeights, ) @@ -477,8 +550,13 @@ async function processSignal(ctx: ProcessContext, signal: IngestSignal): Promise } // CONTRADICT / NEW_TERRITORY — falls through to Sandbox session - } catch { - // Triage failed — fall through to Sandbox session as fallback + } catch (err) { + console.warn("[triage] Triage failed, falling through to sandbox session:", err) + ctx.broadcast("calibrationEvent", { + type: "sessionError", + error: `Triage failed: ${err instanceof Error ? err.message : String(err)}`, + retryable: true, + }) } } @@ -503,21 +581,7 @@ async function processSignal(ctx: ProcessContext, signal: IngestSignal): Promise await ctx.rollbackCheckpoint("session-checkpoint") try { - // Ensure sandbox is running - const { sandbox, sdk } = await ctx.step({ - name: "ensure-sandbox", - timeout: 120_000, - maxRetries: 2, - retryBackoffBase: 5_000, - run: () => ensureSandbox(ctx.state), - }) - - // Mount workspace files - await ctx.step({ - name: "mount-workspace", - timeout: 60_000, - run: () => mountWorkspace(sdk, ctx.state), - }) + const sdk = await withSandbox(ctx) // Run agent session const summary: SessionSummary = await ctx.step({ @@ -542,6 +606,9 @@ async function processSignal(ctx: ProcessContext, signal: IngestSignal): Promise run: () => drainSession({ db: ctx.db, state: ctx.state, broadcast: ctx.broadcast }, summary), }) + // Cache embeddings for new/updated positions + await cachePositionEmbeddings(ctx, [...summary.newPositions, ...summary.updatedPositions]) + // v0.2: Record session cost from sandbox token usage if (summary.tokenUsage && ctx.state.modelPricingCache) { const sessionCostUsd = calculateCost( @@ -570,9 +637,6 @@ async function processSignal(ctx: ProcessContext, signal: IngestSignal): Promise type: "sessionComplete", summary, }) - - // Teardown agent session (not the sandbox — it stays warm) - await teardownSession(sdk, summary.id) } catch (error) { ctx.broadcast("calibrationEvent", { type: "sessionError", @@ -622,7 +686,7 @@ async function logCost( ): Promise { await db.execute( "INSERT INTO cost_log (id, type, cost_usd, input_tokens, output_tokens, created_at) VALUES (?, ?, ?, ?, ?, ?)", - `cost-${Date.now()}-${Math.random().toString(36).slice(2, 6)}`, + `cost-${crypto.randomUUID()}`, type, costUsd, usage?.inputTokens ?? 0, @@ -631,6 +695,88 @@ async function logCost( ) } +// ── Dataset Ingestion (used inside workflow) ───────────────────────── + +async function processDatasetIngestion(ctx: ProcessContext, config: DatasetConfig): Promise { + // Budget check + const costType = needsSandbox(config.source) ? "session" : "triage" + if (!checkBudget(ctx.state, costType)) { + ctx.broadcast("calibrationEvent", { + type: "budgetExhausted", + spentUsd: ctx.state.budgetState.spentTodayUsd, + dailyBudgetUsd: ctx.state.budgetState.dailyBudgetUsd, + }) + return + } + + try { + let output: import("./types").DatasetOutput + + if (needsSandbox(config.source)) { + // Heavy format — run in sandbox + const sdk = await withSandbox(ctx, "dataset") + + output = await ctx.step({ + name: "run-dataset-session", + timeout: 5 * 60 * 1000, + run: () => runDatasetSession(sdk, ctx.state, config), + }) + } else { + // Lightweight format — run on actor + output = await ctx.step({ + name: "ingest-dataset-local", + timeout: 60_000, + run: () => ingestDatasetLocal(config), + }) + } + + // Route output based on mode + if (config.mode === "signals" && output.signals) { + for (const signal of output.signals) { + await ctx.db.execute( + "INSERT INTO signals (id, source_provider, source_mode, content, urgency, credibility) VALUES (?, ?, ?, ?, ?, ?)", + signal.id, + signal.sourceProvider, + signal.sourceMode, + signal.content, + signal.urgency, + signal.credibility, + ) + } + // Process first 50 directly, rest are logged for future tension evaluation + for (const signal of output.signals.slice(0, 50)) { + await processSignal(ctx, signal) + } + } else if (config.mode === "positions" && output.positions) { + await drainDatasetPositions( + { db: ctx.db, state: ctx.state, broadcast: ctx.broadcast }, + output.positions, + output.source, + ) + } else if (config.mode === "analysis" && output.summary) { + if (validateSummary(output.summary)) { + await drainSession( + { db: ctx.db, state: ctx.state, broadcast: ctx.broadcast }, + output.summary, + ) + } + } + + ctx.broadcast("calibrationEvent", { + type: "datasetIngested", + source: output.source, + mode: config.mode, + rowsProcessed: output.rowsProcessed, + }) + } catch (error) { + ctx.broadcast("calibrationEvent", { + type: "datasetError", + source: `${config.source}:${config.uri}`, + error: error instanceof Error ? error.message : String(error), + }) + } +} + // ── Registry ────────────────────────────────────────────────────────── export const registry = setup({ diff --git a/actor/materialize.ts b/actor/materialize.ts index b80a1cc..3a11de8 100644 --- a/actor/materialize.ts +++ b/actor/materialize.ts @@ -80,6 +80,7 @@ export function buildSandboxEnvVars(sandboxConfig: SandboxModelConfig): Record t.length > 4) + .filter((t) => t.length > 5 && !STOP_WORDS.has(t)) + let domainHits = 0 for (const term of domainTerms) { - if (text.includes(term)) return true + if (containsWholeWord(text, term)) domainHits++ } + if (domainHits >= 2) return true // Check existing position slugs as domain vocabulary for (const slug of positionSlugs) { - const slugTerms = slug.split("-").filter((t) => t.length > 3) - const matches = slugTerms.filter((t) => text.includes(t)) + const slugTerms = slug.split("-").filter((t) => t.length > 4 && !STOP_WORDS.has(t)) + const matches = slugTerms.filter((t) => containsWholeWord(text, t)) if (matches.length >= 2) return true } @@ -98,7 +157,7 @@ export async function evaluateTension( positionEmbeddings: Float32Array[], embeddingConfig: ModelRole, signalWeights: Record, -): Promise<{ tension: number; embeddingCost: TokenUsage }> { +): Promise<{ tension: number; embeddingCost: TokenUsage; signalEmbedding?: Float32Array }> { const zeroCost: TokenUsage = { inputTokens: 0, outputTokens: 0 } // Manual /run or owner correction — always trigger @@ -131,5 +190,5 @@ export async function evaluateTension( const sourceWeight = signalWeights[signal.sourceProvider] ?? 1.0 const tension = (1 - maxSimilarity) * sourceWeight - return { tension, embeddingCost: embeddingUsage } + return { tension, embeddingCost: embeddingUsage, signalEmbedding } } diff --git a/actor/types.ts b/actor/types.ts index 09cee93..1cb9939 100644 --- a/actor/types.ts +++ b/actor/types.ts @@ -54,6 +54,39 @@ export interface ModelConfig { sandbox: SandboxModelConfig } +// ── Dataset Config ──────────────────────────────────────────────────── + +export type DatasetSourceType = "huggingface" | "csv" | "parquet" | "json" | "jsonl" | "url" +export type DatasetMode = "signals" | "positions" | "analysis" +export type DatasetSchedule = "init" | "daily" | "manual" + +export interface DatasetConfig { + source: DatasetSourceType + uri: string // HF name, file path, or URL + config?: string // HF dataset config name + split?: string // HF split (train, test, validation) + mode: DatasetMode + limit?: number // max rows (default 1000) + column_map?: Record // source col -> OS field + content_column?: string // which column maps to content/text + label_column?: string // which column maps to confidence/label + credibility?: number // default credibility for signals mode (0.0-1.0) + schedule?: DatasetSchedule // when to run ingestion +} + +export interface DatasetOutput { + source: string + rowsProcessed: number + signals?: IngestSignal[] + positions?: PositionUpdate[] + summary?: SessionSummary +} + +/** Returns true if the dataset source requires the sandbox (Python). */ +export function needsSandbox(source: DatasetSourceType): boolean { + return source === "huggingface" || source === "parquet" +} + // ── Repo Config ─────────────────────────────────────────────────────── export interface RepoConfig { @@ -84,7 +117,7 @@ export interface IngestSignal { // ── Agent Commands ──────────────────────────────────────────────────── export interface AgentCommand { - type: "run" | "correct_position" | "update_config" | "initialize" + type: "run" | "correct_position" | "update_config" | "initialize" | "ingest_dataset" payload: Record } @@ -186,6 +219,8 @@ export type CalibrationEvent = | { type: "budgetExhausted"; spentUsd: number; dailyBudgetUsd: number } | { type: "budgetReset"; dailyBudgetUsd: number } | { type: "triageComplete"; action: TriageAction; slug: string } + | { type: "datasetIngested"; source: string; mode: DatasetMode; rowsProcessed: number } + | { type: "datasetError"; source: string; error: string } // ── Actor State ─────────────────────────────────────────────────────── @@ -207,6 +242,8 @@ export interface SentientState { budgetState: BudgetState triageEnabled: boolean modelPricingCache: ModelPricingCache | null + /** Cached position embeddings keyed by slug. */ + positionEmbeddings: Record } // ── Triage ──────────────────────────────────────────────────────────── @@ -248,6 +285,13 @@ export interface ModelPricingCache { fetchedAt: number // epoch ms } +// ── Actor DB Interface ──────────────────────────────────────────────── + +/** Minimal typed interface for Rivet's opaque DB handle. */ +export interface ActorDb { + execute: (sql: string, ...params: unknown[]) => Promise +} + // ── Skill Metadata ──────────────────────────────────────────────────── export type SkillRuntime = "actor" | "sandbox" @@ -276,6 +320,20 @@ const SandboxModelSchema = z.object({ name: z.string(), }) +const DatasetConfigSchema = z.object({ + source: z.enum(["huggingface", "csv", "parquet", "json", "jsonl", "url"]), + uri: z.string(), + config: z.string().optional(), + split: z.string().optional(), + mode: z.enum(["signals", "positions", "analysis"]), + limit: z.number().optional(), + column_map: z.record(z.string(), z.string()).optional(), + content_column: z.string().optional(), + label_column: z.string().optional(), + credibility: z.number().min(0).max(1).optional(), + schedule: z.enum(["init", "daily", "manual"]).optional(), +}) + const SkillSourceSchema = z.object({ type: z.enum(["github", "local", "git"]), source: z.string(), @@ -335,6 +393,7 @@ export const SentientConfigSchema = z public_positions: z.boolean(), public_inquiries: z.boolean(), }), + datasets: z.array(DatasetConfigSchema).optional(), seed_positions: z .array( z.object({ diff --git a/api/index.ts b/api/index.ts index d4361fb..716db98 100644 --- a/api/index.ts +++ b/api/index.ts @@ -1,3 +1,4 @@ +import { timingSafeEqual } from "node:crypto" import type { Context } from "hono" import { Hono } from "hono" import { handleCalibration } from "./sentient/calibration" @@ -14,7 +15,17 @@ const app = new Hono() const authMiddleware = async (c: Context, next: () => Promise) => { const apiKey = c.req.header("x-api-key") ?? c.req.header("authorization")?.replace("Bearer ", "") if (!apiKey) return c.text("Unauthorized", 401) - // Unkey validation would go here — for now accept any key + + const expected = process.env.API_KEY + if (!expected) return c.text("Server misconfigured: no API_KEY set", 500) + + // Timing-safe comparison to prevent timing attacks + const a = new TextEncoder().encode(apiKey) + const b = new TextEncoder().encode(expected) + if (a.byteLength !== b.byteLength || !timingSafeEqual(a, b)) { + return c.text("Unauthorized", 401) + } + await next() } diff --git a/api/sentient/post.ts b/api/sentient/post.ts index fa991b8..9ab3688 100644 --- a/api/sentient/post.ts +++ b/api/sentient/post.ts @@ -9,7 +9,7 @@ export async function handlePost(c: Context, route: string) { case "signal": { const body = await c.req.json() const signal = { - id: `api-${Date.now()}`, + id: `api-${crypto.randomUUID()}`, sourceProvider: "owner", sourceMode: "api", content: body.content, diff --git a/sandbox/index.ts b/sandbox/index.ts index e0dc485..dc7f5e5 100644 --- a/sandbox/index.ts +++ b/sandbox/index.ts @@ -2,6 +2,8 @@ import { Daytona } from "@daytonaio/sdk" import { SandboxAgent } from "sandbox-agent" import { buildHarnessConfig, buildSandboxEnvVars } from "../actor/materialize" import type { + DatasetConfig, + DatasetOutput, IngestSignal, RepoConfig, SentientState, @@ -27,8 +29,8 @@ export async function ensureSandbox( sandbox = await createFreshSandbox(daytona, state) state.sandboxId = sandbox.id } - } catch { - // Sandbox not found — create fresh + } catch (err) { + console.warn(`[sandbox] Failed to retrieve sandbox ${state.sandboxId}, creating fresh:`, err) sandbox = await createFreshSandbox(daytona, state) state.sandboxId = sandbox.id } @@ -196,7 +198,7 @@ export async function runSession( const summary = JSON.parse(outputText) as SessionSummary // Ensure required fields - summary.id = summary.id || `session-${Date.now()}` + summary.id = summary.id || `session-${crypto.randomUUID()}` summary.startedAt = summary.startedAt || Date.now() - 60_000 summary.completedAt = summary.completedAt || Date.now() summary.triggerSignalId = trigger.id @@ -212,14 +214,107 @@ export async function runSession( return summary } -// ── Teardown Session ────────────────────────────────────────────────── +// ── Dataset Session ────────────────────────────────────────────────── -export async function teardownSession(sdk: SandboxAgent, sessionId: string): Promise { - try { - await sdk.destroySession(sessionId) - } catch { - // Session may already be destroyed +export async function runDatasetSession( + sdk: SandboxAgent, + state: SentientState, + config: DatasetConfig, +): Promise { + const harness = state.modelConfig.sandbox.harness + const model = state.modelConfig.sandbox.name + + // Mount loader.py into sandbox + const loaderPath = "skills/.system/ingest-dataset/loader.py" + const loaderFile = Bun.file(loaderPath) + if (await loaderFile.exists()) { + await sdk.mkdirFs({ path: "/workspace/skills/.system/ingest-dataset" }) + await sdk.writeFsFile( + { path: "/workspace/skills/.system/ingest-dataset/loader.py" }, + await loaderFile.text(), + ) + } + + // Write dataset request config + await sdk.mkdirFs({ path: "/workspace/output" }) + await sdk.writeFsFile( + { path: "/workspace/output/dataset-request.json" }, + JSON.stringify(config, null, 2), + ) + + // Create agent session + const session = await sdk.createSession({ + agent: harness === "opencode" ? "opencode" : harness, + model, + sessionInit: { cwd: "/workspace" }, + }) + + // Build the dataset prompt + const prompt = buildDatasetPrompt(config) + await session.prompt([{ type: "text", text: prompt }]) + + // Determine output file based on mode + const outputFile = + config.mode === "analysis" + ? "/workspace/output/session-summary.json" + : config.mode === "signals" + ? "/workspace/output/dataset-signals.json" + : "/workspace/output/dataset-positions.json" + + // Read output + const outputBytes = await sdk.readFsFile({ path: outputFile }) + const outputText = new TextDecoder().decode(outputBytes) + const output = JSON.parse(outputText) as DatasetOutput + + await sdk.destroySession(session.id) + + return output +} + +function buildDatasetPrompt(config: DatasetConfig): string { + const modeInstructions = { + signals: `Write output to /workspace/output/dataset-signals.json with schema: { source, rowsProcessed, signals: [{ id, sourceProvider: "dataset", sourceMode: "batch", content, urgency: "medium", credibility, timestamp, metadata }] }`, + positions: `Write output to /workspace/output/dataset-positions.json with schema: { source, rowsProcessed, positions: [{ slug, text, confidence, surpriseDelta: 0, priorConfidence: 0, status: "settled" }] }`, + analysis: + "Analyze the dataset using the Alethic Method. Write a standard SessionSummary to /workspace/output/session-summary.json", } + + const loaderArgs = [ + `--source ${config.source}`, + `--uri "${config.uri}"`, + `--mode ${config.mode}`, + `--limit ${config.limit ?? 1000}`, + config.config ? `--config "${config.config}"` : "", + config.split ? `--split "${config.split}"` : "", + config.content_column ? `--content-col "${config.content_column}"` : "", + config.label_column ? `--label-col "${config.label_column}"` : "", + config.credibility !== undefined ? `--credibility ${config.credibility}` : "", + config.column_map ? `--column-map '${JSON.stringify(config.column_map)}'` : "", + ] + .filter(Boolean) + .join(" ") + + const outputFilename = config.mode === "analysis" ? "session-summary" : `dataset-${config.mode}` + + return [ + "# Dataset Ingestion Task", + "", + "Read the dataset request at /workspace/output/dataset-request.json.", + "", + "Use the Python loader script to process the dataset:", + "```bash", + `python3 /workspace/skills/.system/ingest-dataset/loader.py ${loaderArgs} > /workspace/output/${outputFilename}.json`, + "```", + "", + "Or use Python directly if the loader doesn't fit your needs.", + "", + "## Output Requirements", + modeInstructions[config.mode], + "", + config.mode === "analysis" + ? "Apply domain reasoning to the dataset contents. Identify positions, contradictions, and inquiries." + : `Transform each row into the output schema. Use content_column="${config.content_column ?? "auto-detect"}" for the main text field.`, + ].join("\n") } // ── Context Building ────────────────────────────────────────────────── diff --git a/sentient.jsonc b/sentient.jsonc index eea38a3..acc7850 100644 --- a/sentient.jsonc +++ b/sentient.jsonc @@ -93,6 +93,20 @@ // "url": "https://os.archive.energy" // }, + // ── Datasets ────────────────────────────────────────────────────── + // Source-agnostic data ingestion. Lightweight formats (csv, json, jsonl, url) + // run on the actor; heavy formats (huggingface, parquet) run in the sandbox. + // "datasets": [ + // { "source": "huggingface", "uri": "user/domain-data", "split": "train", + // "mode": "signals", "limit": 500, "content_column": "text", + // "credibility": 0.7, "schedule": "init" }, + // { "source": "csv", "uri": "knowledge/seed-data.csv", + // "mode": "positions", "content_column": "claim", + // "label_column": "confidence" }, + // { "source": "url", "uri": "https://api.example.com/data.json", + // "mode": "signals", "schedule": "daily", "limit": 100 } + // ], + // ── Seed State ──────────────────────────────────────────────────── "seed_positions": [ { "slug": "initial-position", "text": "Initial position about the domain", "confidence": 0.8 } diff --git a/skills/.system/ingest-dataset/SKILL.md b/skills/.system/ingest-dataset/SKILL.md new file mode 100644 index 0000000..7c09241 --- /dev/null +++ b/skills/.system/ingest-dataset/SKILL.md @@ -0,0 +1,69 @@ +--- +name: ingest-dataset +description: Generic dataset ingestion with pluggable source adapters. + Supports HuggingFace, CSV, Parquet, JSON/JSONL, and URL endpoints. + Lightweight formats run on the actor; heavy formats run in the sandbox. +runtime: actor +escalate_to: sandbox +trigger: ingest_dataset command or datasets config in sentient.jsonc +--- + +# Ingest — Dataset + +Transforms structured data from multiple sources into signals or +positions for the knowledge graph. + +## Source Adapters + +| Source | Runtime | Library | +|--------|---------|---------| +| `huggingface` | Sandbox | `datasets` (Python) | +| `parquet` | Sandbox | `pyarrow` / `pandas` (Python) | +| `csv` | Actor | `Bun.csv()` native | +| `json` / `jsonl` | Actor | `Bun.file().json()` | +| `url` | Actor | `fetch` + auto-detect format | + +## Output Modes + +- **signals** — Each row becomes an `IngestSignal` flowing through the tension pipeline. Output: `dataset-signals.json` +- **positions** — Each row becomes a position written directly to knowledge/. Output: `dataset-positions.json` +- **analysis** — Sandbox agent loads dataset and applies Alethic Method, producing a standard `SessionSummary`. Output: `session-summary.json` + +## Sandbox Execution (HuggingFace / Parquet) + +```bash +python3 /workspace/skills/.system/ingest-dataset/loader.py \ + --source huggingface --uri "squad" --split "validation" \ + --mode signals --limit 100 --content-col "question" --credibility 0.7 \ + > /workspace/output/dataset-signals.json +``` + +## Column Mapping + +Use `column_map` to rename source columns: + +```json +{ "column_map": { "question": "content", "answer_score": "confidence", "id": "slug" } } +``` + +Special fields: `content_column` (main text), `label_column` (credibility/confidence). + +## Configuration + +```jsonc +"datasets": [ + { "source": "huggingface", "uri": "user/domain-data", "split": "train", + "mode": "signals", "limit": 500, "content_column": "text", + "credibility": 0.7, "schedule": "init" } +] +``` + +## Schedule + +- `init` — Run once during Sentient initialization +- `daily` — Run during daily scan cycle +- `manual` — Only via API (`POST /sentient/run` with dataset config) + +## Environment + +`HF_TOKEN` — Required for private HuggingFace datasets (optional for public). diff --git a/skills/.system/ingest-dataset/loader.py b/skills/.system/ingest-dataset/loader.py new file mode 100644 index 0000000..8cecf83 --- /dev/null +++ b/skills/.system/ingest-dataset/loader.py @@ -0,0 +1,166 @@ +#!/usr/bin/env python3 +""" +Dataset loader for sandbox execution (HuggingFace / Parquet). + +Usage: + python3 loader.py --source huggingface --uri "squad" --mode signals [options] + +Lightweight formats (csv, json, jsonl, url) are handled actor-side in TypeScript. +This loader only handles formats that require Python: huggingface and parquet. + +Output: JSON to stdout matching the requested mode schema. +""" + +import argparse +import json +import sys +import time + + +def load_huggingface(uri, config=None, split=None, limit=1000): + from datasets import load_dataset + + kwargs = {} + if config: + kwargs["name"] = config + if split: + kwargs["split"] = split + + ds = load_dataset(uri, **kwargs, streaming=True) + rows = [] + for i, row in enumerate(ds): + if i >= limit: + break + rows.append({k: v for k, v in row.items() if isinstance(v, (str, int, float, bool))}) + return rows + + +def load_parquet(uri, limit=1000): + import pandas as pd + + df = pd.read_parquet(uri).head(limit) + return df.to_dict(orient="records") + + +LOADERS = { + "huggingface": load_huggingface, + "parquet": load_parquet, +} + + +def extract_content(row, content_col=None): + if content_col and content_col in row: + return str(row[content_col]) + for key in ("text", "content", "body", "description", "question"): + if key in row and isinstance(row[key], str): + return row[key] + first_str = next((v for v in row.values() if isinstance(v, str)), None) + return first_str if first_str else json.dumps(row) + + +def extract_confidence(row, label_col=None, default=0.5): + if label_col and label_col in row: + try: + val = float(row[label_col]) + return max(0.0, min(1.0, val)) + except (ValueError, TypeError): + pass + return default + + +def apply_column_map(row, column_map): + if not column_map: + return row + mapped = {} + for src, dest in column_map.items(): + if src in row: + mapped[dest] = row[src] + for key, val in row.items(): + if key not in column_map: + mapped[key] = val + return mapped + + +def to_signals(rows, args): + now = int(time.time() * 1000) + signals = [] + for i, row in enumerate(rows): + row = apply_column_map(row, args.column_map) + signals.append({ + "id": f"dataset-{args.uri}-{i}-{now}", + "sourceProvider": "dataset", + "sourceMode": "batch", + "content": extract_content(row, args.content_col), + "urgency": "medium", + "credibility": extract_confidence(row, args.label_col, args.credibility), + "timestamp": now, + "metadata": { + "datasetSource": args.source, + "datasetUri": args.uri, + "rowIndex": i, + }, + }) + return {"source": f"{args.source}:{args.uri}", "rowsProcessed": len(rows), "signals": signals} + + +def to_positions(rows, args): + positions = [] + uri_slug = "".join(c if c.isalnum() else "-" for c in args.uri).strip("-") + for i, row in enumerate(rows): + row = apply_column_map(row, args.column_map) + slug = str(row.get("slug", f"dataset-{uri_slug}-{i}")) + positions.append({ + "slug": slug, + "text": extract_content(row, args.content_col), + "confidence": extract_confidence(row, args.label_col, args.credibility), + "surpriseDelta": 0, + "priorConfidence": 0, + "status": "settled", + }) + return {"source": f"{args.source}:{args.uri}", "rowsProcessed": len(rows), "positions": positions} + + +def main(): + parser = argparse.ArgumentParser(description="Dataset loader for OpenSentient") + parser.add_argument("--source", required=True, choices=LOADERS.keys()) + parser.add_argument("--uri", required=True) + parser.add_argument("--config", default=None, help="HF dataset config name") + parser.add_argument("--split", default=None, help="HF dataset split") + parser.add_argument("--mode", required=True, choices=["signals", "positions", "analysis"]) + parser.add_argument("--limit", type=int, default=1000) + parser.add_argument("--content-col", default=None) + parser.add_argument("--label-col", default=None) + parser.add_argument("--credibility", type=float, default=0.5) + parser.add_argument("--column-map", default=None, help="JSON string of column mappings") + args = parser.parse_args() + + if args.column_map: + args.column_map = json.loads(args.column_map) + + loader = LOADERS[args.source] + loader_kwargs = {"uri": args.uri, "limit": args.limit} + if args.source == "huggingface": + loader_kwargs["config"] = args.config + loader_kwargs["split"] = args.split + + rows = loader(**loader_kwargs) + + if args.mode == "signals": + output = to_signals(rows, args) + elif args.mode == "positions": + output = to_positions(rows, args) + elif args.mode == "analysis": + output = { + "source": f"{args.source}:{args.uri}", + "rowsProcessed": len(rows), + "rows": rows, + } + else: + print(f"Unknown mode: {args.mode}", file=sys.stderr) + sys.exit(1) + + json.dump(output, sys.stdout, indent=2, default=str) + + +if __name__ == "__main__": + main()