From 25f822325339703036752ee04e1793edbca170c7 Mon Sep 17 00:00:00 2001 From: john Date: Sat, 27 Jun 2026 08:25:02 +0800 Subject: [PATCH] refactor: Business logic and dependencies updates MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 核心服务代码更新 (db, server, auth, proxy) - Agent 相关模块更新 (mindspace, experience) - 前端组件和 hooks 更新 - 数据库 schema 更新 - 依赖版本更新 --- db.mjs | 49 ++++- experience-service-pg.mjs | 227 +++++++++++++++++++++ experience-service.mjs | 181 +++++++++++++++++ experience-service.test.mjs | 114 +++++++++++ mindspace-agent-jobs.mjs | 167 +++++++++++----- mindspace-agent-runner.mjs | 53 ++++- mindspace-assets.mjs | 53 ++++- mindspace-publications.mjs | 116 ++++++++++- package.json | 1 + schema.sql | 34 ++++ server.mjs | 261 ++++++++++++++++++++++++- src/App.tsx | 3 + src/api/client.ts | 36 +++- src/components/MindSpacePageDetail.tsx | 78 +++++++- src/components/ShareSheet.tsx | 36 +++- src/hooks/useTKMindChat.ts | 35 +--- src/types.ts | 1 + src/utils/sessionStorage.ts | 39 ++++ tkmind-proxy.mjs | 57 +++++- user-auth.mjs | 33 +++- 20 files changed, 1458 insertions(+), 116 deletions(-) create mode 100644 experience-service-pg.mjs create mode 100644 experience-service.mjs create mode 100644 experience-service.test.mjs create mode 100644 src/utils/sessionStorage.ts diff --git a/db.mjs b/db.mjs index 92dedcc..27cebe2 100644 --- a/db.mjs +++ b/db.mjs @@ -13,13 +13,27 @@ export function isDatabaseConfigured() { ); } +// Pool size must scale with peak concurrent agent sessions: each session does +// several queries per turn (session node lookup + policy + provider + reconcile). +// The historical default of 10 starved at ~30 concurrent sessions. Tune via +// MYSQL_POOL_SIZE per portal instance × expected concurrency. +function resolvePoolOptions() { + const connectionLimit = Math.max(1, Number(process.env.MYSQL_POOL_SIZE ?? 50)); + // queueLimit caps how many requests wait for a free connection before failing + // fast, instead of piling up unboundedly under a connection-starvation storm. + const queueLimit = Math.max(0, Number(process.env.MYSQL_POOL_QUEUE_LIMIT ?? 0)); + return { waitForConnections: true, connectionLimit, queueLimit }; +} + export function createDbPool() { if (!isDatabaseConfigured()) { throw new Error('MySQL 未配置，请设置 DATABASE_URL 或 MYSQL_* 环境变量'); } + const poolOptions = resolvePoolOptions(); + if (process.env.DATABASE_URL) { - return mysql.createPool(process.env.DATABASE_URL); + return mysql.createPool({ uri: process.env.DATABASE_URL, ...poolOptions }); } return mysql.createPool({ @@ -28,8 +42,7 @@ export function createDbPool() { user: process.env.MYSQL_USER ?? 'boot', password: process.env.MYSQL_PASSWORD ?? '', database: process.env.MYSQL_DATABASE ?? 'tkmind', - waitForConnections: true, - connectionLimit: 10, + ...poolOptions, }); } @@ -199,6 +212,26 @@ export async function migrateSchema(pool) { ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci `); + // Shared experience store (etat C). Keep in sync with schema.sql. + await pool.query(` + CREATE TABLE IF NOT EXISTS h5_experience ( + id CHAR(36) PRIMARY KEY, + scope VARCHAR(64) NOT NULL DEFAULT 'global', + kind VARCHAR(32) NOT NULL DEFAULT 'lesson', + title VARCHAR(255) NOT NULL, + body MEDIUMTEXT NOT NULL, + tags_json JSON NULL, + source_session_id VARCHAR(128) NULL, + source_user_id CHAR(36) NULL, + use_count BIGINT NOT NULL DEFAULT 0, + embedding JSON NULL, + created_at BIGINT NOT NULL, + updated_at BIGINT NOT NULL, + KEY idx_h5_experience_scope_updated (scope, updated_at), + FULLTEXT KEY ftx_h5_experience (title, body) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci + `); + const publishColumns = [ ['plaza_view_count', 'BIGINT NOT NULL DEFAULT 0'], ['plaza_like_count', 'BIGINT NOT NULL DEFAULT 0'], @@ -239,6 +272,16 @@ export async function migrateSchema(pool) { ); } + // goosed_target stores the resolved upstream URL a session is pinned to, so + // routing survives reordering or resizing the goosed target list. The legacy + // integer goosed_node is kept as a fallback for rows written before this column + // existed (and for older bundles that still read it). + if (!(await columnExists(pool, 'h5_user_sessions', 'goosed_target'))) { + await pool.query( + `ALTER TABLE h5_user_sessions ADD COLUMN goosed_target VARCHAR(255) NULL AFTER goosed_node`, + ); + } + const oauthStateColumns = [ ['intent', "VARCHAR(16) NOT NULL DEFAULT 'login' AFTER utm_campaign"], ['bind_user_id', 'CHAR(36) NULL AFTER intent'], diff --git a/experience-service-pg.mjs b/experience-service-pg.mjs new file mode 100644 index 0000000..360bcef --- /dev/null +++ b/experience-service-pg.mjs @@ -0,0 +1,227 @@ +import crypto from 'node:crypto'; + +/** + * PostgreSQL + pgvector backed experience store (etat C, polyglot mode). + * + * Same contract as the MySQL `createExperienceService` (record / search / + * reflect) so callers (mindspace-agent-runner) are backend-agnostic. The MySQL + * business DB is untouched; only this shared-experience workload lives on PG. + * + * Semantic search via pgvector is used when an `embed(text) => number[]` function + * is supplied; otherwise it degrades to keyword (ILIKE) + recency, identical in + * spirit to the MySQL version, so it is useful before an embedding provider is wired. + * + * The `pg` driver is imported dynamically so this module loads even when the + * dependency is not yet installed; it is only required once the service is built. + * + * @param {object} options + * @param {string} options.connectionString - e.g. process.env.EXPERIENCE_PG_URL + * @param {(text: string) => Promise} [options.embed] - embedding fn + * @param {number} [options.embeddingDim=1536] - vector dimension (must match embed) + * @param {() => number} [options.now] + * @param {number} [options.recencyHalfLifeMs] + */ +export async function createPgExperienceService(options = {}) { + const connectionString = options.connectionString; + if (!connectionString) { + throw new Error('createPgExperienceService 需要 connectionString（建议用 EXPERIENCE_PG_URL）'); + } + const now = options.now ?? (() => Date.now()); + const embed = options.embed ?? null; + const embeddingDim = Math.max(1, Number(options.embeddingDim ?? 1536)); + const halfLifeMs = Math.max( + 60_000, + Number(options.recencyHalfLifeMs ?? 30 * 24 * 60 * 60 * 1000), + ); + + const { default: pg } = await import('pg'); + const pool = new pg.Pool({ connectionString, max: Number(options.poolMax ?? 10) }); + + // Idempotent setup. pgvector must be available on the server (CREATE EXTENSION). + await pool.query('CREATE EXTENSION IF NOT EXISTS vector'); + await pool.query(` + CREATE TABLE IF NOT EXISTS h5_experience ( + id UUID PRIMARY KEY, + scope TEXT NOT NULL DEFAULT 'global', + kind TEXT NOT NULL DEFAULT 'lesson', + title TEXT NOT NULL, + body TEXT NOT NULL, + tags JSONB NOT NULL DEFAULT '[]'::jsonb, + source_session_id TEXT, + source_user_id TEXT, + use_count BIGINT NOT NULL DEFAULT 0, + embedding vector(${embeddingDim}), + created_at BIGINT NOT NULL, + updated_at BIGINT NOT NULL + ) + `); + await pool.query( + 'CREATE INDEX IF NOT EXISTS idx_h5_experience_scope_updated ON h5_experience (scope, updated_at DESC)', + ); + // Trigram-ish keyword fallback index (btree on lower(title)) is cheap; the + // ANN vector index is created lazily once rows have embeddings. + await pool.query( + "CREATE INDEX IF NOT EXISTS idx_h5_experience_embedding ON h5_experience USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100)", + ).catch(() => { + // ivfflat needs the extension + may warn on empty table; non-fatal. + }); + + function normalizeTags(tags) { + if (!Array.isArray(tags)) return []; + return [...new Set(tags.map((t) => String(t).trim()).filter(Boolean))]; + } + + function toVectorLiteral(vec) { + // pgvector accepts a string like '[0.1,0.2,...]' + return `[${vec.map((n) => Number(n)).join(',')}]`; + } + + function rowToExperience(row) { + return { + id: row.id, + scope: row.scope, + kind: row.kind, + title: row.title, + body: row.body, + tags: Array.isArray(row.tags) ? row.tags : [], + sourceSessionId: row.source_session_id ?? null, + sourceUserId: row.source_user_id ?? null, + useCount: Number(row.use_count ?? 0), + createdAt: Number(row.created_at), + updatedAt: Number(row.updated_at), + }; + } + + async function record(input) { + const title = String(input?.title ?? '').trim(); + const body = String(input?.body ?? '').trim(); + if (!title) throw experienceError('经验标题不能为空', 'invalid_experience_input'); + if (!body) throw experienceError('经验内容不能为空', 'invalid_experience_input'); + const id = crypto.randomUUID(); + const ts = now(); + const scope = String(input?.scope ?? 'global').trim() || 'global'; + const kind = String(input?.kind ?? 'lesson').trim() || 'lesson'; + const tags = normalizeTags(input?.tags); + let embedding = null; + if (embed) { + try { + const vec = await embed(`${title}\n${body}`); + if (Array.isArray(vec) && vec.length === embeddingDim) embedding = toVectorLiteral(vec); + } catch { + embedding = null; // embedding failure must not block recording + } + } + await pool.query( + `INSERT INTO h5_experience + (id, scope, kind, title, body, tags, source_session_id, source_user_id, + use_count, embedding, created_at, updated_at) + VALUES ($1,$2,$3,$4,$5,$6::jsonb,$7,$8,0,$9::vector,$10,$11)`, + [ + id, + scope, + kind, + title, + body, + JSON.stringify(tags), + input?.sourceSessionId ?? null, + input?.sourceUserId ?? null, + embedding, + ts, + ts, + ], + ); + return rowToExperience({ + id, + scope, + kind, + title, + body, + tags, + source_session_id: input?.sourceSessionId ?? null, + source_user_id: input?.sourceUserId ?? null, + use_count: 0, + created_at: ts, + updated_at: ts, + }); + } + + async function search(query, { scope = 'global', limit = 5 } = {}) { + const text = String(query ?? '').trim(); + if (!text) return []; + const max = Math.max(1, limit); + + // Vector path: cosine distance ordering, recency as tiebreaker. + if (embed) { + try { + const vec = await embed(text); + if (Array.isArray(vec) && vec.length === embeddingDim) { + const { rows } = await pool.query( + `SELECT id, scope, kind, title, body, tags, source_session_id, + source_user_id, use_count, created_at, updated_at + FROM h5_experience + WHERE scope = $1 AND embedding IS NOT NULL + ORDER BY embedding <=> $2::vector, updated_at DESC + LIMIT $3`, + [scope, toVectorLiteral(vec), max], + ); + if (rows.length > 0) { + await bumpUseCount(rows.map((r) => r.id)); + return rows.map(rowToExperience); + } + } + } catch { + // fall through to keyword path + } + } + + // Keyword fallback: ILIKE any term, ordered by recency. + const terms = tokenize(text); + if (terms.length === 0) return []; + const likeClauses = terms.map((_, i) => `(title ILIKE $${i + 2} OR body ILIKE $${i + 2})`); + const params = [scope, ...terms.map((t) => `%${t}%`), max]; + const { rows } = await pool.query( + `SELECT id, scope, kind, title, body, tags, source_session_id, + source_user_id, use_count, created_at, updated_at + FROM h5_experience + WHERE scope = $1 AND (${likeClauses.join(' OR ')}) + ORDER BY updated_at DESC + LIMIT $${terms.length + 2}`, + params, + ); + if (rows.length > 0) await bumpUseCount(rows.map((r) => r.id)); + return rows.map(rowToExperience); + } + + async function bumpUseCount(ids) { + if (!ids.length) return; + await pool + .query('UPDATE h5_experience SET use_count = use_count + 1 WHERE id = ANY($1::uuid[])', [ids]) + .catch(() => {}); + } + + function tokenize(query) { + return [ + ...new Set( + String(query ?? '') + .toLowerCase() + .split(/[^\p{L}\p{N}]+/u) + .map((t) => t.trim()) + .filter((t) => t.length >= 2), + ), + ]; + } + + async function reflect() { + return { ok: false, reason: 'not_implemented' }; + } + + async function close() { + await pool.end(); + } + + return { record, search, reflect, close }; +} + +function experienceError(message, code) { + return Object.assign(new Error(message), { code }); +} diff --git a/experience-service.mjs b/experience-service.mjs new file mode 100644 index 0000000..5e4c3eb --- /dev/null +++ b/experience-service.mjs @@ -0,0 +1,181 @@ +import crypto from 'node:crypto'; + +/** + * Shared experience store (etat C of the goose scale plan). + * + * All goosed Worker instances read/write the same learned experience so that + * capability is not siloed per instance. This first implementation is backed by + * MySQL (`h5_experience`) with a keyword + recency ranking. The retrieval scoring + * is deliberately isolated in `rankRows` so that swapping to PostgreSQL + pgvector + * later only replaces the query/scoring layer, not the calling contract + * (record / search / reflect). + * + * @param {import('mysql2/promise').Pool} pool + * @param {object} [options] + * @param {() => number} [options.now] - injectable clock for tests + */ +export function createExperienceService(pool, options = {}) { + const now = options.now ?? (() => Date.now()); + // Recency half-life: a record this old contributes half its keyword score. + const halfLifeMs = Math.max( + 60_000, + Number(options.recencyHalfLifeMs ?? 30 * 24 * 60 * 60 * 1000), + ); + + function normalizeTags(tags) { + if (!Array.isArray(tags)) return []; + return [...new Set(tags.map((t) => String(t).trim()).filter(Boolean))]; + } + + function rowToExperience(row) { + return { + id: row.id, + scope: row.scope, + kind: row.kind, + title: row.title, + body: row.body, + tags: row.tags_json ? safeJsonArray(row.tags_json) : [], + sourceSessionId: row.source_session_id ?? null, + sourceUserId: row.source_user_id ?? null, + useCount: Number(row.use_count ?? 0), + createdAt: Number(row.created_at), + updatedAt: Number(row.updated_at), + }; + } + + function safeJsonArray(value) { + if (Array.isArray(value)) return value; + try { + const parsed = JSON.parse(value); + return Array.isArray(parsed) ? parsed : []; + } catch { + return []; + } + } + + // Keyword overlap × recency decay. Kept separate so the pgvector backend can + // replace this with cosine similarity without touching search()'s shape. + function rankRows(rows, queryTerms, nowMs) { + const terms = queryTerms; + return rows + .map((row) => { + const haystack = `${row.title}\n${row.body}`.toLowerCase(); + let keywordScore = 0; + for (const term of terms) { + if (haystack.includes(term)) keywordScore += 1; + } + const ageMs = Math.max(0, nowMs - Number(row.updated_at)); + const recency = Math.pow(0.5, ageMs / halfLifeMs); + const score = keywordScore * recency; + return { row, score }; + }) + .filter((entry) => entry.score > 0) + .sort((a, b) => b.score - a.score); + } + + function tokenize(query) { + return [ + ...new Set( + String(query ?? '') + .toLowerCase() + .split(/[^\p{L}\p{N}]+/u) + .map((t) => t.trim()) + .filter((t) => t.length >= 2), + ), + ]; + } + + /** + * Persist a piece of experience. Returns the stored record. + */ + async function record(input) { + const title = String(input?.title ?? '').trim(); + const body = String(input?.body ?? '').trim(); + if (!title) throw experienceError('经验标题不能为空', 'invalid_experience_input'); + if (!body) throw experienceError('经验内容不能为空', 'invalid_experience_input'); + const id = crypto.randomUUID(); + const ts = now(); + const scope = String(input?.scope ?? 'global').trim() || 'global'; + const kind = String(input?.kind ?? 'lesson').trim() || 'lesson'; + const tags = normalizeTags(input?.tags); + await pool.query( + `INSERT INTO h5_experience + (id, scope, kind, title, body, tags_json, source_session_id, source_user_id, + use_count, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?)`, + [ + id, + scope, + kind, + title, + body, + JSON.stringify(tags), + input?.sourceSessionId ?? null, + input?.sourceUserId ?? null, + ts, + ts, + ], + ); + return rowToExperience({ + id, + scope, + kind, + title, + body, + tags_json: tags, + source_session_id: input?.sourceSessionId ?? null, + source_user_id: input?.sourceUserId ?? null, + use_count: 0, + created_at: ts, + updated_at: ts, + }); + } + + /** + * Retrieve the most relevant experience for a query within a scope. + * @returns {Promise} ranked experiences (best first), max `limit`. + */ + async function search(query, { scope = 'global', limit = 5 } = {}) { + const terms = tokenize(query); + if (terms.length === 0) return []; + // Pull a bounded candidate set by scope+recency, then rank in JS. The + // candidate cap keeps this cheap on MySQL; pgvector would push ranking into SQL. + const [rows] = await pool.query( + `SELECT id, scope, kind, title, body, tags_json, source_session_id, + source_user_id, use_count, created_at, updated_at + FROM h5_experience + WHERE scope = ? + ORDER BY updated_at DESC + LIMIT 500`, + [scope], + ); + const ranked = rankRows(rows, terms, now()).slice(0, Math.max(1, limit)); + if (ranked.length > 0) { + // Best-effort usage bump so popular experience can be surfaced/weighted later. + const ids = ranked.map((entry) => entry.row.id); + const placeholders = ids.map(() => '?').join(','); + await pool + .query( + `UPDATE h5_experience SET use_count = use_count + 1 WHERE id IN (${placeholders})`, + ids, + ) + .catch(() => {}); + } + return ranked.map((entry) => rowToExperience(entry.row)); + } + + /** + * Placeholder for the reflection pass that distills raw session traces into + * durable lessons. Wired later once the record() trigger points are defined; + * kept here so the calling contract is stable. + */ + async function reflect() { + return { ok: false, reason: 'not_implemented' }; + } + + return { record, search, reflect }; +} + +function experienceError(message, code) { + return Object.assign(new Error(message), { code }); +} diff --git a/experience-service.test.mjs b/experience-service.test.mjs new file mode 100644 index 0000000..06836f2 --- /dev/null +++ b/experience-service.test.mjs @@ -0,0 +1,114 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { createExperienceService } from './experience-service.mjs'; + +// Minimal in-memory pool covering the SQL shapes experience-service issues: +// INSERT, scope-filtered SELECT, and the use_count bump UPDATE. +function createMockPool() { + const rows = []; + const query = async (sql, params = []) => { + if (sql.includes('INSERT INTO h5_experience')) { + const [ + id, + scope, + kind, + title, + body, + tagsJson, + sourceSessionId, + sourceUserId, + createdAt, + updatedAt, + ] = params; + rows.push({ + id, + scope, + kind, + title, + body, + tags_json: tagsJson, + source_session_id: sourceSessionId, + source_user_id: sourceUserId, + use_count: 0, + created_at: createdAt, + updated_at: updatedAt, + }); + return [{ affectedRows: 1 }]; + } + if (sql.includes('FROM h5_experience') && sql.includes('WHERE scope = ?')) { + const [scope] = params; + const matched = rows + .filter((r) => r.scope === scope) + .sort((a, b) => b.updated_at - a.updated_at); + return [matched.map((r) => ({ ...r }))]; + } + if (sql.includes('UPDATE h5_experience SET use_count')) { + for (const id of params) { + const row = rows.find((r) => r.id === id); + if (row) row.use_count += 1; + } + return [{ affectedRows: params.length }]; + } + throw new Error(`unexpected SQL: ${sql}`); + }; + return { pool: { query }, rows }; +} + +test('record persists an experience with normalized fields', async () => { + const { pool, rows } = createMockPool(); + const svc = createExperienceService(pool, { now: () => 1000 }); + const saved = await svc.record({ + title: 'Headscale 部署', + body: '在 103 上用固定公网 IP 部署 headscale，注意防火墙放行 41641/udp。', + tags: ['headscale', 'headscale', ' 网络 ', ''], + }); + assert.equal(rows.length, 1); + assert.equal(saved.scope, 'global'); + assert.equal(saved.kind, 'lesson'); + assert.deepEqual(saved.tags, ['headscale', '网络']); + assert.equal(saved.useCount, 0); +}); + +test('record rejects empty title or body', async () => { + const { pool } = createMockPool(); + const svc = createExperienceService(pool); + await assert.rejects(() => svc.record({ title: '', body: 'x' }), /标题不能为空/); + await assert.rejects(() => svc.record({ title: 'x', body: '' }), /内容不能为空/); +}); + +test('search ranks keyword matches and ignores empty queries', async () => { + const { pool } = createMockPool(); + const svc = createExperienceService(pool, { now: () => 1_000_000 }); + await svc.record({ title: 'Headscale 部署', body: 'headscale 防火墙 udp' }); + await svc.record({ title: '无关经验', body: '这条跟查询完全无关的内容' }); + + assert.deepEqual(await svc.search(' '), []); + + const hits = await svc.search('headscale 部署'); + assert.equal(hits.length, 1); + assert.equal(hits[0].title, 'Headscale 部署'); +}); + +test('search applies recency decay so newer wins on equal keyword score', async () => { + const { pool } = createMockPool(); + let clock = 0; + const svc = createExperienceService(pool, { + now: () => clock, + recencyHalfLifeMs: 1000, + }); + clock = 0; + await svc.record({ title: '部署指南 A', body: 'deploy 部署 指南' }); + clock = 10_000; // 10 half-lives newer + await svc.record({ title: '部署指南 B', body: 'deploy 部署 指南' }); + clock = 10_000; + const hits = await svc.search('部署 deploy 指南', { limit: 2 }); + assert.equal(hits[0].title, '部署指南 B'); +}); + +test('search bumps use_count for returned rows', async () => { + const { pool, rows } = createMockPool(); + const svc = createExperienceService(pool, { now: () => 5 }); + await svc.record({ title: 'Caddy 灰度', body: 'caddy weighted_round_robin 分流' }); + await svc.search('caddy 灰度'); + assert.equal(rows[0].use_count, 1); +}); diff --git a/mindspace-agent-jobs.mjs b/mindspace-agent-jobs.mjs index 467c95a..8ad0fed 100644 --- a/mindspace-agent-jobs.mjs +++ b/mindspace-agent-jobs.mjs @@ -442,6 +442,96 @@ export function createAgentJobService(pool, options = {}) { } }; + // Marks a locked, queued job row as running and returns the claim payload. + // Caller must hold a row lock on `job` inside `conn`'s open transaction; this + // commits the transaction. + const finalizeClaim = async (conn, job) => { + const jobId = job.id; + const token = crypto.randomBytes(24).toString('base64url'); + const now = nowFactory(); + await conn.query( + `UPDATE h5_agent_jobs + SET status = 'running', started_at = COALESCE(started_at, ?), heartbeat_at = ?, + updated_at = ?, expires_at = ?, progress_json = ?, job_token_hash = ? + WHERE id = ?`, + [ + now, + now, + now, + now + tokenTtlMs, + JSON.stringify({ stage: 'preparing_files' }), + hashJobToken(token), + jobId, + ], + ); + await conn.commit(); + const assets = await fetchJobAssets(jobId); + const permissionScope = jsonValue(job.permission_scope, {}); + const userContext = jsonValue(job.user_context_json, { + locale: 'zh-CN', + timezone: 'Asia/Shanghai', + }); + return { + jobId, + userId: job.user_id, + jobToken: token, + instruction: job.instruction, + userContext, + allowedAssets: assets.map((asset) => ({ + assetId: asset.assetId, + versionId: asset.assetVersionId, + permission: asset.permission, + displayName: asset.displayName, + mimeType: asset.mimeType, + downloadEndpoint: `/api/internal/agent/jobs/${jobId}/assets/${asset.assetId}`, + })), + output: { + categoryId: job.output_category_id, + categoryCode: job.output_category_code, + allowedTypes: [job.output_type], + maxBytes: asNumber(job.max_output_bytes), + }, + capabilities: permissionScope.capabilities ?? { + network: false, + shell: false, + createPage: true, + }, + expiresAt: now + tokenTtlMs, + }; + }; + + // Atomically claim the oldest queued, unexpired job. Uses SKIP LOCKED so that + // multiple worker loops (across portal instances) pull distinct jobs without + // blocking each other. Returns null when the queue is empty. + const claimNextJob = async () => { + const conn = await pool.getConnection(); + try { + await conn.beginTransaction(); + const [rows] = await conn.query( + `SELECT j.*, c.category_code AS output_category_code + FROM h5_agent_jobs j + JOIN h5_space_categories c ON c.id = j.output_category_id + WHERE j.status = 'queued' + AND (j.expires_at IS NULL OR j.expires_at >= ?) + ORDER BY j.queued_at ASC + LIMIT 1 + FOR UPDATE SKIP LOCKED`, + [nowFactory()], + ); + const job = rows[0]; + if (!job) { + await conn.commit(); + return null; + } + return await finalizeClaim(conn, job); + } catch (error) { + await conn.rollback(); + throw error; + } finally { + conn.release(); + } + }; + const claimJob = async (jobId) => { const conn = await pool.getConnection(); try { @@ -463,57 +553,7 @@ export function createAgentJobService(pool, options = {}) { if (job.expires_at != null && asNumber(job.expires_at) < nowFactory()) { throw agentJobError('任务已过期', 'agent_job_expired'); } - const token = crypto.randomBytes(24).toString('base64url'); - const now = nowFactory(); - await conn.query( - `UPDATE h5_agent_jobs - SET status = 'running', started_at = COALESCE(started_at, ?), heartbeat_at = ?, - updated_at = ?, expires_at = ?, progress_json = ?, job_token_hash = ? - WHERE id = ?`, - [ - now, - now, - now, - now + tokenTtlMs, - JSON.stringify({ stage: 'preparing_files' }), - hashJobToken(token), - jobId, - ], - ); - await conn.commit(); - const assets = await fetchJobAssets(jobId); - const permissionScope = jsonValue(job.permission_scope, {}); - const userContext = jsonValue(job.user_context_json, { - locale: 'zh-CN', - timezone: 'Asia/Shanghai', - }); - return { - jobId, - userId: job.user_id, - jobToken: token, - instruction: job.instruction, - userContext, - allowedAssets: assets.map((asset) => ({ - assetId: asset.assetId, - versionId: asset.assetVersionId, - permission: asset.permission, - displayName: asset.displayName, - mimeType: asset.mimeType, - downloadEndpoint: `/api/internal/agent/jobs/${jobId}/assets/${asset.assetId}`, - })), - output: { - categoryId: job.output_category_id, - categoryCode: job.output_category_code, - allowedTypes: [job.output_type], - maxBytes: asNumber(job.max_output_bytes), - }, - capabilities: permissionScope.capabilities ?? { - network: false, - shell: false, - createPage: true, - }, - expiresAt: now + tokenTtlMs, - }; + return await finalizeClaim(conn, job); } catch (error) { await conn.rollback(); throw error; @@ -522,6 +562,7 @@ export function createAgentJobService(pool, options = {}) { } }; + const requireJobToken = async (jobId, token) => { const [rows] = await pool.query( `SELECT j.*, c.category_code AS output_category_code @@ -648,6 +689,28 @@ export function createAgentJobService(pool, options = {}) { return getJob(job.user_id, jobId); }; + // Restart resilience: a job left in 'running' by a crashed/restarted worker + // stops heartbeating. Mark such jobs failed with the retryable worker_crashed + // code so they surface to the user (and can be retried) instead of being stuck + // forever. We fail rather than auto-requeue to avoid poison-job loops, since + // there is no per-job attempt counter. + const reapStaleJobs = async (maxStaleMs = 5 * 60 * 1000) => { + const now = nowFactory(); + const cutoff = now - Math.max(1, maxStaleMs); + const [result] = await pool.query( + `UPDATE h5_agent_jobs + SET status = 'failed', error_code = 'worker_crashed', + error_message = 'worker heartbeat timed out', + completed_at = ?, updated_at = ?, heartbeat_at = NULL, job_token_hash = NULL, + progress_json = ? + WHERE status = 'running' + AND heartbeat_at IS NOT NULL + AND heartbeat_at < ?`, + [now, now, JSON.stringify({ stage: 'failed' }), cutoff], + ); + return result?.affectedRows ?? 0; + }; + return { createJob, getJob, @@ -655,6 +718,8 @@ export function createAgentJobService(pool, options = {}) { cancelJob, retryJob, claimJob, + claimNextJob, + reapStaleJobs, getAssetForJob, heartbeat, completeJob, diff --git a/mindspace-agent-runner.mjs b/mindspace-agent-runner.mjs index 05fbd12..bba3a43 100644 --- a/mindspace-agent-runner.mjs +++ b/mindspace-agent-runner.mjs @@ -305,6 +305,7 @@ export function createMindSpaceAgentRunner({ apiSecret, userAuth, agentJobService, + experienceService = null, executeSessionReply = defaultExecuteSessionReply, apiFetchImpl = null, }) { @@ -324,11 +325,15 @@ export function createMindSpaceAgentRunner({ }); }); - const runJob = async (jobId) => { + // preClaimed lets a polling worker that already atomically claimed the job + // (via claimNextJob + SKIP LOCKED) hand the claim payload straight in, instead + // of claiming a second time. The HTTP-triggered path calls runJob(jobId) with + // no preClaimed and claims here as before. + const runJob = async (jobId, preClaimed = null) => { let claim = null; let sessionId = null; try { - claim = await agentJobService.claimJob(jobId); + claim = preClaimed ?? (await agentJobService.claimJob(jobId)); const gate = await userAuth.canUseChat(claim.userId); if (!gate.ok) { throw runnerError(gate.message || '当前用户无法执行 Agent 任务', 'worker_unavailable'); @@ -377,7 +382,11 @@ export function createMindSpaceAgentRunner({ const localAsset = await agentJobService.getAssetForJob(jobId, claim.jobToken, asset.assetId); assetContexts.push(await readAssetContext(localAsset)); } - const prompt = buildAgentJobPrompt(claim, assetContexts); + let prompt = buildAgentJobPrompt(claim, assetContexts); + // Inject shared experience (etat C): relevant lessons all instances learned. + // Best-effort — retrieval must never block or fail a job. + const relevant = await retrieveExperience(claim.instruction); + if (relevant) prompt = `${relevant}\n\n${prompt}`; const requestId = crypto.randomUUID(); const reply = await executeSessionReply(apiFetch, sessionId, requestId, prompt); const parsed = normalizeStructuredResult(extractJsonObject(reply.text)); @@ -386,7 +395,7 @@ export function createMindSpaceAgentRunner({ await userAuth.billSessionUsage(claim.userId, sessionId, reply.tokenState, requestId); } - return agentJobService.completeJob(jobId, claim.jobToken, { + const completed = await agentJobService.completeJob(jobId, claim.jobToken, { title: parsed.title, summary: parsed.summary, content: parsed.content, @@ -395,6 +404,9 @@ export function createMindSpaceAgentRunner({ templateId: parsed.templateId, sourceAssetIds: claim.allowedAssets.map((asset) => asset.assetId), }); + // Record the outcome as shared experience so other instances benefit. + await recordExperience(claim, sessionId, parsed); + return completed; } catch (error) { if (claim?.jobToken) { await agentJobService @@ -409,6 +421,39 @@ export function createMindSpaceAgentRunner({ } }; + // Returns a prompt-ready block of relevant prior experience, or '' if none / + // disabled / on any error. Never throws. + async function retrieveExperience(instruction) { + if (!experienceService || !instruction) return ''; + try { + const hits = await experienceService.search(instruction, { limit: 3 }); + if (!hits.length) return ''; + const lines = hits.map((h) => `- ${h.title}: ${h.body}`).join('\n'); + return `# 相关经验（供参考，来自历史任务）\n${lines}`; + } catch { + return ''; + } + } + + // Persists a one-line lesson from a completed job. Best-effort; never throws. + async function recordExperience(claim, sessionId, parsed) { + if (!experienceService) return; + try { + const title = String(parsed?.title ?? claim.instruction ?? '').slice(0, 200); + const summary = String(parsed?.summary ?? '').trim(); + if (!title || !summary) return; + await experienceService.record({ + kind: 'task_outcome', + title, + body: summary, + sourceSessionId: sessionId, + sourceUserId: claim.userId, + }); + } catch { + // swallow — experience recording is non-critical + } + } + return { runJob }; } diff --git a/mindspace-assets.mjs b/mindspace-assets.mjs index 2905494..55dd42c 100644 --- a/mindspace-assets.mjs +++ b/mindspace-assets.mjs @@ -32,7 +32,7 @@ const ALLOWED_EXTENSIONS = new Map([ ['.html', 'text/html'], ['.htm', 'text/html'], ]); -const MAX_IMAGE_UPLOAD_BYTES = 1536 * 1024; +const MAX_IMAGE_UPLOAD_BYTES = 2 * 1024 * 1024; const PUBLIC_TEMP_IMAGE_DIR = '.tmp-images'; const PUBLIC_IMAGE_EXTENSIONS = new Map([ @@ -65,6 +65,54 @@ function expectedMimeType(filename) { return ALLOWED_EXTENSIONS.get(path.extname(filename).toLowerCase()) ?? null; } +function extractImageDimensions(buffer, mimeType) { + if (mimeType === 'image/png' && buffer.length >= 24) { + const width = buffer.readUInt32BE(16); + const height = buffer.readUInt32BE(20); + return { width, height }; + } + if (mimeType === 'image/jpeg' && buffer.length >= 2) { + let offset = 2; + while (offset < buffer.length - 9) { + if (buffer[offset] !== 0xff) break; + const marker = buffer[offset + 1]; + if (marker === 0xd9) break; + if (marker >= 0xd0 && marker <= 0xd8) { + offset += 2; + continue; + } + const segmentLength = buffer.readUInt16BE(offset + 2); + if (marker === 0xc0 || marker === 0xc1 || marker === 0xc2) { + const height = buffer.readUInt16BE(offset + 5); + const width = buffer.readUInt16BE(offset + 7); + return { width, height }; + } + offset += segmentLength + 2; + } + } + if (mimeType === 'image/webp' && buffer.length >= 30) { + const width = buffer.readUInt32LE(24) + 1; + const height = buffer.readUInt32LE(28) + 1; + return { width, height }; + } + return null; +} + +function validateImagePixels(buffer, mimeType) { + if (!mimeType?.startsWith('image/')) return null; + const dims = extractImageDimensions(buffer, mimeType); + if (!dims) return { ok: true }; + const { width, height } = dims; + const megapixels = (width * height) / 1_000_000; + if (megapixels > 25) { + throw Object.assign( + new Error(`图片像素超过限制（${megapixels.toFixed(1)}MP > 25MP）`), + { code: 'image_pixel_exceeded' } + ); + } + return { ok: true, width, height }; +} + function detectMimeType(buffer, filename) { const head = buffer.subarray(0, 256).toString('utf8').trimStart().toLowerCase(); if (head.startsWith(' MAX_IMAGE_UPLOAD_BYTES) { throw Object.assign(new Error('图片文件超过单文件大小限制'), { code: 'file_too_large' }); } + if (detectedMimeType.startsWith('image/')) { + validateImagePixels(buffer, detectedMimeType); + } const target = absoluteStoragePath(upload.temporary_storage_key); await fs.mkdir(path.dirname(target), { recursive: true }); diff --git a/mindspace-publications.mjs b/mindspace-publications.mjs index c6bc567..61d7500 100644 --- a/mindspace-publications.mjs +++ b/mindspace-publications.mjs @@ -6,6 +6,7 @@ import { replacePrivateResourceReferences, scanContent } from './mindspace-conte import { pageInternals } from './mindspace-pages.mjs'; import { loadMindSpaceConfig } from './mindspace-config.mjs'; import { resolvePublicBaseUrl } from './user-publish.mjs'; +import { createImgproxySigner } from './imgproxy-signer.mjs'; const SCANNER_VERSION = 'mindspace-content-v1'; const PRIVATE_ASSET_DOWNLOAD_URL_PATTERN = @@ -159,7 +160,13 @@ function buildPublicationThumbnailFallback(ownerSlug, urlSlug) { return `/u/${encodeURIComponent(ownerSlug)}/pages/${encodeURIComponent(urlSlug)}.thumbnail.png`; } -async function localizePrivateImageReferences({ pool, userId, html, absoluteStoragePath }) { +async function localizePrivateImageReferences({ + pool, + userId, + html, + absoluteStoragePath, + imgproxySigner = null, +}) { const source = String(html ?? ''); const matches = [...source.matchAll(PRIVATE_ASSET_DOWNLOAD_URL_PATTERN)]; if (matches.length === 0) return source; @@ -181,9 +188,14 @@ async function localizePrivateImageReferences({ pool, userId, html, absoluteStor for (const assetId of assetIds) { const asset = byId.get(assetId); if (!asset) continue; - const mimeType = String(asset.mime_type || 'application/octet-stream'); - const buffer = await fs.readFile(absoluteStoragePath(asset.storage_key)); - replacements.set(assetId, `data:${mimeType};base64,${buffer.toString('base64')}`); + + if (imgproxySigner) { + replacements.set(assetId, imgproxySigner.buildUrl(imgproxySigner.baseUrl, asset.storage_key, 'display')); + } else { + const mimeType = String(asset.mime_type || 'application/octet-stream'); + const buffer = await fs.readFile(absoluteStoragePath(asset.storage_key)); + replacements.set(assetId, `data:${mimeType};base64,${buffer.toString('base64')}`); + } } if (replacements.size === 0) return source; @@ -199,6 +211,7 @@ async function prepareHtmlPublishContent({ ownerSlug, urlSlug, absoluteStoragePath, + imgproxySigner = null, }) { let publishContent = (await localizeGoogleFontsCss(html)).html; publishContent = await localizePrivateImageReferences({ @@ -206,6 +219,7 @@ async function prepareHtmlPublishContent({ userId, html: publishContent, absoluteStoragePath, + imgproxySigner, }); return replacePrivateResourceReferences( publishContent, @@ -217,6 +231,21 @@ export function createPublicationService(pool, options = {}) { const storageRoot = path.resolve(options.storageRoot ?? path.join(process.cwd(), 'data', 'mindspace')); const idFactory = options.idFactory ?? (() => crypto.randomUUID()); const publicPageLimitFallback = Number(options.publicPageLimit ?? 5); + + let imgproxySigner = null; + if (process.env.IMGPROXY_BASE_URL && process.env.IMGPROXY_SIGNING_KEY && process.env.IMGPROXY_SIGNING_SALT) { + try { + imgproxySigner = createImgproxySigner( + process.env.IMGPROXY_SIGNING_KEY, + process.env.IMGPROXY_SIGNING_SALT, + ); + imgproxySigner.baseUrl = process.env.IMGPROXY_BASE_URL; + console.log('[Publication] imgproxy signer initialized'); + } catch (err) { + console.warn('[Publication] imgproxy signer init failed:', err instanceof Error ? err.message : err); + } + } + const resolvePublicPageLimit = async () => { try { const config = await loadMindSpaceConfig(pool); @@ -309,6 +338,7 @@ export function createPublicationService(pool, options = {}) { ownerSlug, urlSlug, absoluteStoragePath, + imgproxySigner: imgproxySigner ? { buildUrl: (path, preset) => imgproxySigner.buildUrl(imgproxySigner.baseUrl, path, preset) } : null, }); }; @@ -549,9 +579,9 @@ export function createPublicationService(pool, options = {}) { await conn.query( `INSERT INTO h5_publish_records (id, user_id, page_id, page_version_id, publish_type, url_slug, public_url, - access_mode, password_hash, token_hash, token_prefix, expires_at, published_at, + access_mode, password_hash, token_hash, token_prefix, expires_at, user_confirmed_at, published_at, status, view_count, security_scan_id, created_at, updated_at) - VALUES (?, ?, ?, ?, 'page', ?, ?, ?, ?, ?, ?, ?, ?, 'online', 0, ?, ?, ?)`, + VALUES (?, ?, ?, ?, 'page', ?, ?, ?, ?, ?, ?, ?, ?, ?, 'online', 0, ?, ?, ?)`, [ publishId, userId, @@ -564,6 +594,7 @@ export function createPublicationService(pool, options = {}) { tokenHash, privateToken?.slice(0, 8) ?? null, result.expiresAt, + null, now, scanId, now, @@ -605,6 +636,23 @@ export function createPublicationService(pool, options = {}) { WHERE id = ? AND user_id = ?`, [htmlBytes, now, page.space_id, userId], ); + + const assetUrlPattern = /\/api\/mindspace\/v1\/assets\/([a-z0-9-]+)/gi; + const assetIds = new Set(); + let match; + while ((match = assetUrlPattern.exec(html)) !== null) { + assetIds.add(match[1]); + } + if (assetIds.size > 0) { + const refValues = Array.from(assetIds).map((assetId) => [publishId, assetId, now]); + await conn.query( + `INSERT IGNORE INTO h5_publication_asset_refs + (publication_id, asset_id, created_at) VALUES ` + + refValues.map(() => '(?, ?, ?)').join(','), + refValues.flat(), + ); + } + await conn.commit(); return { ...publicationResponse({ @@ -642,6 +690,46 @@ export function createPublicationService(pool, options = {}) { return publicationResponse(rows[0]); }; + const updatePublicationStatus = async (userId, publicationId, { accessMode, expiresAt }) => { + const conn = await pool.getConnection(); + try { + await conn.beginTransaction(); + const [rows] = await conn.query( + `SELECT * FROM h5_publish_records + WHERE id = ? AND user_id = ? AND status = 'online' LIMIT 1 FOR UPDATE`, + [publicationId, userId], + ); + const publication = rows[0]; + if (!publication) throw publicationError('发布记录不存在', 'publication_not_found'); + + const normalizedMode = normalizeAccessMode(accessMode); + const normalizedExpiresAt = normalizeExpiresAt(expiresAt, normalizedMode === 'time_limited'); + const now = Date.now(); + + await conn.query( + `UPDATE h5_publish_records + SET access_mode = ?, expires_at = ?, user_confirmed_at = ?, updated_at = ? + WHERE id = ? AND user_id = ?`, + [normalizedMode, normalizedExpiresAt, now, now, publicationId, userId], + ); + await conn.commit(); + + const updated = { + ...publication, + access_mode: normalizedMode, + expires_at: normalizedExpiresAt, + user_confirmed_at: now, + updated_at: now, + }; + return publicationResponse(updated); + } catch (error) { + await conn.rollback(); + throw error; + } finally { + conn.release(); + } + }; + const offline = async (userId, publicationId) => { const conn = await pool.getConnection(); try { @@ -832,6 +920,20 @@ export function createPublicationService(pool, options = {}) { return publicHomepageResponse(owner, rows); }; + const cleanupExpiredUnconfirmedPublications = async (now = Date.now()) => { + const [result] = await pool.query( + `UPDATE h5_publish_records + SET access_mode = 'private', expires_at = NULL, updated_at = ? + WHERE access_mode = 'public' + AND expires_at IS NOT NULL + AND expires_at <= ? + AND user_confirmed_at IS NULL + AND status = 'online'`, + [now, now], + ); + return { cleaned: result.affectedRows }; + }; + return { check, publish, @@ -839,8 +941,10 @@ export function createPublicationService(pool, options = {}) { getPublicHomepage, getStats, offline, + updatePublicationStatus, resolvePublic, resolvePrivateLink, + cleanupExpiredUnconfirmedPublications, }; } diff --git a/package.json b/package.json index 98f9065..3a963fe 100644 --- a/package.json +++ b/package.json @@ -57,6 +57,7 @@ "http-proxy-middleware": "^3.0.3", "jsonrepair": "^3.14.0", "mysql2": "^3.22.5", + "pg": "^8.22.0", "qrcode": "^1.5.4", "react": "^19.0.0", "react-dom": "^19.0.0", diff --git a/schema.sql b/schema.sql index c02695e..b1ca566 100644 --- a/schema.sql +++ b/schema.sql @@ -223,6 +223,7 @@ CREATE TABLE IF NOT EXISTS h5_publish_records ( token_hash CHAR(64) NULL, token_prefix VARCHAR(16) NULL, expires_at BIGINT NULL, + user_confirmed_at BIGINT NULL, published_at BIGINT NOT NULL, offline_at BIGINT NULL, status ENUM('draft', 'online', 'expired', 'offline', 'blocked') NOT NULL DEFAULT 'online', @@ -232,6 +233,7 @@ CREATE TABLE IF NOT EXISTS h5_publish_records ( updated_at BIGINT NOT NULL, KEY idx_h5_publish_route (url_slug, status, published_at), KEY idx_h5_publish_page (user_id, page_id, published_at), + KEY idx_h5_publish_auto_private (expires_at, user_confirmed_at) WHERE access_mode = 'public' AND user_confirmed_at IS NULL, UNIQUE KEY uq_h5_publish_token_hash (token_hash), CONSTRAINT fk_h5_publish_user FOREIGN KEY (user_id) REFERENCES h5_users(id) ON DELETE CASCADE, CONSTRAINT fk_h5_publish_page FOREIGN KEY (page_id) REFERENCES h5_page_records(id) ON DELETE CASCADE, @@ -239,6 +241,17 @@ CREATE TABLE IF NOT EXISTS h5_publish_records ( CONSTRAINT fk_h5_publish_scan FOREIGN KEY (security_scan_id) REFERENCES h5_security_scans(id) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; +CREATE TABLE IF NOT EXISTS h5_publication_asset_refs ( + publication_id CHAR(36) NOT NULL, + asset_id CHAR(36) NOT NULL, + created_at BIGINT NOT NULL, + PRIMARY KEY (publication_id, asset_id), + KEY idx_asset_id (asset_id), + KEY idx_created_at (created_at), + CONSTRAINT fk_h5_pub_asset_ref_pub FOREIGN KEY (publication_id) REFERENCES h5_publish_records(id) ON DELETE CASCADE, + CONSTRAINT fk_h5_pub_asset_ref_asset FOREIGN KEY (asset_id) REFERENCES h5_assets(id) ON DELETE CASCADE +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + CREATE TABLE IF NOT EXISTS h5_publication_events ( id CHAR(36) PRIMARY KEY, publish_id CHAR(36) NOT NULL, @@ -997,3 +1010,24 @@ CREATE TABLE IF NOT EXISTS h5_blocked_words ( updated_at BIGINT NOT NULL, UNIQUE KEY uk_blocked_word (word) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +-- Shared experience store (etat C): all goosed Worker instances read/write the +-- same learned experience so capability is not siloed per instance. Built on +-- MySQL first with a keyword + recency retrieval; `embedding` is reserved for a +-- later move to PostgreSQL + pgvector without changing the calling contract. +CREATE TABLE IF NOT EXISTS h5_experience ( + id CHAR(36) PRIMARY KEY, + scope VARCHAR(64) NOT NULL DEFAULT 'global', + kind VARCHAR(32) NOT NULL DEFAULT 'lesson', + title VARCHAR(255) NOT NULL, + body MEDIUMTEXT NOT NULL, + tags_json JSON NULL, + source_session_id VARCHAR(128) NULL, + source_user_id CHAR(36) NULL, + use_count BIGINT NOT NULL DEFAULT 0, + embedding JSON NULL, + created_at BIGINT NOT NULL, + updated_at BIGINT NOT NULL, + KEY idx_h5_experience_scope_updated (scope, updated_at), + FULLTEXT KEY ftx_h5_experience (title, body) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; diff --git a/server.mjs b/server.mjs index 7f430f7..c619976 100644 --- a/server.mjs +++ b/server.mjs @@ -95,6 +95,7 @@ import { createScheduleService } from './schedule-service.mjs'; import { startScheduleReminderWorker } from './schedule-reminder-worker.mjs'; import { createLlmProviderService, RELAY_BOOTSTRAP } from './llm-providers.mjs'; import { createSessionSnapshotService } from './session-snapshot.mjs'; +import { createExperienceService } from './experience-service.mjs'; import { attachAsrRoutes } from './asr-proxy.mjs'; import { isNativeH5ApiPath } from './policies.mjs'; @@ -186,6 +187,10 @@ const rawUploadBody = express.raw({ type: 'application/octet-stream', limit: Number(process.env.MINDSPACE_MAX_FILE_BYTES ?? DEFAULT_MAX_FILE_BYTES), }); +const rawUploadBodyImage = express.raw({ + type: 'application/octet-stream', + limit: 2 * 1024 * 1024, +}); const wikiAuth = createWikiAuth(path.join(__dirname, PUBLISH_ROOT_DIR, 'wiki-db')); @@ -270,6 +275,16 @@ async function bootstrapUserAuth() { process.env.MINDSPACE_STORAGE_ROOT ?? path.join(__dirname, 'data', 'mindspace'), publicPageLimit: Number(process.env.MINDSPACE_FREE_PUBLIC_PAGE_LIMIT ?? 5), }); + setInterval(async () => { + try { + const result = await mindSpacePublications.cleanupExpiredUnconfirmedPublications(); + if (result.cleaned > 0) { + console.log(`[Publication Cleanup] Auto-privatized ${result.cleaned} expired unconfirmed publications`); + } + } catch (err) { + console.error('[Publication Cleanup Error]', err instanceof Error ? err.message : err); + } + }, 60 * 1000); await ensureAlgorithmConfig(pool); const plazaAlgorithmConfig = await loadAlgorithmConfig(pool); plazaRedis = await createPlazaRedis(process.env.PLAZA_REDIS_URL, pool); @@ -347,13 +362,104 @@ async function bootstrapUserAuth() { process.env.MINDSPACE_STORAGE_ROOT ?? path.join(__dirname, 'data', 'mindspace'), maxOutputBytes: Number(process.env.MINDSPACE_MAX_FILE_BYTES ?? DEFAULT_MAX_FILE_BYTES), }); + // Shared experience store (etat C): retrieval before / recording after each + // agent job, so all instances learn from one another. Gated so it can be + // disabled without touching the runner. Polyglot: when EXPERIENCE_PG_URL is + // set we use PostgreSQL + pgvector (semantic search) for this workload only; + // the MySQL business DB is untouched. Falls back to MySQL keyword store if PG + // init fails (e.g. driver missing) so a misconfig never blocks startup. + let experienceService = null; + if (process.env.MINDSPACE_EXPERIENCE_ENABLED !== 'false') { + if (process.env.EXPERIENCE_PG_URL) { + try { + const { createPgExperienceService } = await import('./experience-service-pg.mjs'); + experienceService = await createPgExperienceService({ + connectionString: process.env.EXPERIENCE_PG_URL, + }); + console.log('Experience store: PostgreSQL + pgvector'); + } catch (error) { + console.error( + 'Experience PG init failed, falling back to MySQL store:', + error instanceof Error ? error.message : error, + ); + experienceService = createExperienceService(pool); + } + } else { + experienceService = createExperienceService(pool); + } + } mindSpaceAgentRunner = createMindSpaceAgentRunner({ apiTarget: API_TARGET, apiSecret: API_SECRET, userAuth, agentJobService: mindSpaceAgentJobs, + experienceService, }); mindSpaceAudit = createMindSpaceAuditWriter(pool); + // Agent job consumer: a DB-polling worker that atomically claims queued jobs + // (claimNextJob uses SELECT ... FOR UPDATE SKIP LOCKED, so multiple instances + // can run this loop without double-processing) and runs them via the runner. + // Opt-in per instance: must NOT run on the 105 stateless front (see + // docs/g2-load-balancing.md) — gate with MINDSPACE_AGENT_WORKER_ENABLED. + if (process.env.MINDSPACE_AGENT_WORKER_ENABLED === 'true') { + const workerConcurrency = Math.max( + 1, + Number(process.env.MINDSPACE_AGENT_WORKER_CONCURRENCY ?? 2), + ); + const workerPollMs = Math.max( + 200, + Number(process.env.MINDSPACE_AGENT_WORKER_POLL_MS ?? 1000), + ); + const workerStaleMs = Math.max( + 10_000, + Number(process.env.MINDSPACE_AGENT_WORKER_STALE_MS ?? 5 * 60 * 1000), + ); + let inFlight = 0; + let draining = false; + const drainQueue = async () => { + if (draining) return; + draining = true; + try { + while (inFlight < workerConcurrency) { + const claim = await mindSpaceAgentJobs.claimNextJob(); + if (!claim) break; + inFlight += 1; + void mindSpaceAgentRunner + .runJob(claim.jobId, claim) + .catch((error) => { + console.error('Agent worker job failed:', error); + }) + .finally(() => { + inFlight -= 1; + }); + } + } catch (error) { + console.error('Agent worker drain failed:', error); + } finally { + draining = false; + } + }; + const workerTimer = setInterval(() => { + void drainQueue(); + }, workerPollMs); + const reaperTimer = setInterval(() => { + void mindSpaceAgentJobs + .reapStaleJobs(workerStaleMs) + .then((reaped) => { + if (reaped > 0) { + console.warn(`Agent worker reaped ${reaped} stale running job(s)`); + } + }) + .catch((error) => { + console.error('Agent worker reaper failed:', error); + }); + }, Math.min(workerStaleMs, 60_000)); + workerTimer.unref?.(); + reaperTimer.unref?.(); + console.log( + `Agent job worker enabled (concurrency=${workerConcurrency}, poll=${workerPollMs}ms)`, + ); + } if (WORKSPACE_MAINTENANCE_ENABLED) { startWorkspaceThumbnailWatcher(path.join(__dirname, PUBLISH_ROOT_DIR)); startWorkspaceAssetSyncWatcher({ @@ -1742,6 +1848,75 @@ api.get('/mindspace/v1/agent/jobs/:jobId', async (req, res) => { } }); +// Server-Sent Events stream of a job's progress, so long-running agent tasks can +// be dispatched async (enqueue → 202 → subscribe here) instead of holding a +// synchronous streaming connection. Polls the job (ownership enforced by getJob) +// and pushes on change; closes on terminal status or client disconnect. +api.get('/mindspace/v1/agent/jobs/:jobId/stream', async (req, res) => { + if (!mindSpaceAgentJobs || !ensureMindSpaceEnabled(res, req, { agent: true })) return; + const TERMINAL = new Set(['completed', 'failed', 'cancelled', 'timed_out']); + let job; + try { + job = await mindSpaceAgentJobs.getJob(req.currentUser.id, req.params.jobId); + } catch (error) { + return mindSpaceError(res, req, error); + } + res.writeHead(200, { + 'Content-Type': 'text/event-stream', + 'Cache-Control': 'no-cache, no-transform', + Connection: 'keep-alive', + 'X-Accel-Buffering': 'no', + }); + const send = (event, payload) => { + res.write(`event: ${event}\n`); + res.write(`data: ${JSON.stringify(payload)}\n\n`); + }; + let lastSignature = ''; + const emitIfChanged = (current) => { + const signature = `${current.status}:${JSON.stringify(current.progress ?? {})}`; + if (signature !== lastSignature) { + lastSignature = signature; + send('progress', current); + } + return signature; + }; + emitIfChanged(job); + if (TERMINAL.has(job.status)) { + send('done', job); + return res.end(); + } + let closed = false; + const cleanup = () => { + if (closed) return; + closed = true; + clearInterval(pollTimer); + clearInterval(keepAliveTimer); + }; + const pollTimer = setInterval(async () => { + if (closed) return; + try { + const current = await mindSpaceAgentJobs.getJob(req.currentUser.id, req.params.jobId); + emitIfChanged(current); + if (TERMINAL.has(current.status)) { + send('done', current); + cleanup(); + res.end(); + } + } catch { + // Job vanished or transient read error: end the stream rather than leak it. + cleanup(); + res.end(); + } + }, Math.max(500, Number(process.env.MINDSPACE_AGENT_SSE_POLL_MS ?? 1000))); + // Comment line keeps proxies from closing an idle connection. + const keepAliveTimer = setInterval(() => { + if (!closed) res.write(': keep-alive\n\n'); + }, 15_000); + pollTimer.unref?.(); + keepAliveTimer.unref?.(); + req.on('close', cleanup); +}); + api.get('/mindspace/v1/agent/jobs', async (req, res) => { if (!mindSpaceAgentJobs || !ensureMindSpaceEnabled(res, req, { agent: true })) return; try { @@ -1893,7 +2068,7 @@ api.post('/mindspace/v1/uploads', async (req, res) => { } }); -api.put('/mindspace/v1/uploads/:uploadId/content', rawUploadBody, async (req, res) => { +api.put('/mindspace/v1/uploads/:uploadId/content', rawUploadBodyImage, async (req, res) => { if (!mindSpaceAssets) return res.status(503).json({ message: 'MindSpace 未启用' }); try { const result = await mindSpaceAssets.writeUploadContent( @@ -1947,6 +2122,58 @@ api.get('/mindspace/v1/assets', async (req, res) => { } }); +api.get('/mindspace/v1/authorize-image', async (req, res) => { + if (!mindSpacePublications) { + return res.status(503).json({ error: 'Service unavailable' }); + } + const assetId = String(req.query.asset_id ?? ''); + if (!assetId) { + return res.status(400).json({ error: 'Missing asset_id parameter' }); + } + try { + const [refs] = await pool.query( + `SELECT pr.access_mode, pr.expires_at + FROM h5_publication_asset_refs refs + JOIN h5_publish_records pr ON refs.publication_id = pr.id + WHERE refs.asset_id = ? AND pr.status = 'online' + ORDER BY CASE + WHEN pr.access_mode = 'public' THEN 0 + WHEN pr.access_mode = 'time_limited' THEN 1 + ELSE 2 + END, + pr.expires_at DESC + LIMIT 1`, + [assetId], + ); + + if (!refs[0]) { + return res.status(403).json({ error: 'Forbidden' }); + } + + const publication = refs[0]; + const now = Date.now(); + + if (publication.access_mode === 'public') { + res.set('Cache-Control', 'public, max-age=31536000, immutable'); + return res.status(200).json({ ok: true }); + } + + if ( + publication.access_mode === 'time_limited' && + publication.expires_at && + Number(publication.expires_at) > now + ) { + res.set('Cache-Control', 'public, max-age=60'); + return res.status(200).json({ ok: true }); + } + + return res.status(403).json({ error: 'Forbidden' }); + } catch (error) { + console.error('[authorize-image]', error instanceof Error ? error.message : error); + return res.status(500).json({ error: 'Internal server error' }); + } +}); + api.get('/mindspace/v1/assets/:assetId/download', async (req, res) => { if (!mindSpaceAssets || !ensureMindSpaceEnabled(res, req)) return; try { @@ -2865,6 +3092,30 @@ api.post('/mindspace/v1/pages/:pageId/publish', async (req, res) => { } }); +api.post('/mindspace/v1/publications/:publicationId/update-status', async (req, res) => { + if (!mindSpacePublications) return res.status(503).json({ message: 'MindSpace 未启用' }); + try { + const publication = await mindSpacePublications.updatePublicationStatus( + req.currentUser.id, + req.params.publicationId, + { + accessMode: req.body?.access_mode, + expiresAt: req.body?.expires_at, + }, + ); + await mindSpaceAudit?.write({ + userId: req.currentUser.id, + action: 'publication.status_updated', + objectType: 'publication', + objectId: req.params.publicationId, + ip: req.ip, + }); + return sendData(res, req, publication); + } catch (error) { + return mindSpaceError(res, req, error); + } +}); + api.post('/mindspace/v1/publications/:publicationId/offline', async (req, res) => { if (!mindSpacePublications) return res.status(503).json({ message: 'MindSpace 未启用' }); try { @@ -3290,9 +3541,11 @@ api.get('/sessions/:sessionId', async (req, res, next) => { if (sessionSnapshotService?.isEnabled()) { const messages = (gooseSession.conversation ?? []) .filter((m) => m.metadata?.userVisible); - void sessionSnapshotService - .save(sessionId, req.currentUser.id, gooseSession, messages) - .catch(() => {}); + if (messages.length > 0) { + void sessionSnapshotService + .save(sessionId, req.currentUser.id, gooseSession, messages) + .catch(() => {}); + } } return res.json(gooseSession); } catch (err) { diff --git a/src/App.tsx b/src/App.tsx index 7a516d4..81dcbfc 100644 --- a/src/App.tsx +++ b/src/App.tsx @@ -1,6 +1,7 @@ import { useEffect, useState } from 'react'; import { Navigate, Route, Routes, useLocation, useNavigate } from 'react-router-dom'; import { checkAuth, logout, setUnauthorizedHandler } from './api/client'; +import { clearAllStoredSessionIds } from './utils/sessionStorage'; import { loadBlockedWords } from './utils/wordFilter'; import { AuthView } from './components/AuthView'; import { ChatView } from './components/ChatView'; @@ -51,6 +52,7 @@ function AuthenticatedApp({ const navigate = useNavigate(); const handleLogout = () => { + clearAllStoredSessionIds(); void logout().finally(() => { onLogout(); navigate('/', { replace: true }); @@ -191,6 +193,7 @@ export function App() { grantedSkills={grantedSkills} onUserUpdate={setUser} onLogout={() => { + clearAllStoredSessionIds(); setAuthed(false); setUser(null); }} diff --git a/src/api/client.ts b/src/api/client.ts index 70ad5ec..00e30e4 100644 --- a/src/api/client.ts +++ b/src/api/client.ts @@ -995,6 +995,26 @@ export async function publishMindSpacePage( return result.data; } +export async function updatePublicationStatus( + publicationId: string, + input: { + accessMode: MindSpacePublishCheck['accessMode']; + expiresAt?: number | null; + }, +): Promise { + const result = await apiFetch<{ data: MindSpacePublication }>( + `/mindspace/v1/publications/${encodeURIComponent(publicationId)}/update-status`, + { + method: 'POST', + body: JSON.stringify({ + access_mode: input.accessMode, + expires_at: input.expiresAt, + }), + }, + ); + return result.data; +} + export async function redactMindSpacePage( pageId: string, input: { @@ -1967,12 +1987,16 @@ export async function listSessions(): Promise { return result.sessions ?? []; } +function sessionPath(sessionId: string, suffix = '') { + return `/sessions/${encodeURIComponent(sessionId)}${suffix}`; +} + export async function getSession(sessionId: string): Promise { - return apiFetch(`/sessions/${sessionId}`); + return apiFetch(sessionPath(sessionId)); } export async function deleteChatSession(sessionId: string): Promise { - await apiFetch(`/sessions/${sessionId}`, { method: 'DELETE' }); + await apiFetch(sessionPath(sessionId), { method: 'DELETE' }); } export async function loadSessionDetail( @@ -1986,7 +2010,7 @@ export async function loadSessionDetail( if (hints?.messageCount != null) params.set('hint_mc', String(hints.messageCount)); if (hints?.updatedAt) params.set('hint_ua', hints.updatedAt); const qs = params.size ? `?${params.toString()}` : ''; - const detail = await apiFetch(`/sessions/${sessionId}${qs}`); + const detail = await apiFetch(`${sessionPath(sessionId)}${qs}`); const messages = normalizeConversationMessages( (detail.conversation ?? []).filter((m) => m.metadata?.userVisible), ); @@ -1998,7 +2022,7 @@ export async function sendReply( requestId: string, userMessage: Message, ): Promise { - await apiFetch(`/sessions/${sessionId}/reply`, { + await apiFetch(`${sessionPath(sessionId)}/reply`, { method: 'POST', body: JSON.stringify({ request_id: requestId, @@ -2008,7 +2032,7 @@ export async function sendReply( } export async function cancelRequest(sessionId: string, requestId: string): Promise { - await apiFetch(`/sessions/${sessionId}/cancel`, { + await apiFetch(`${sessionPath(sessionId)}/cancel`, { method: 'POST', body: JSON.stringify({ request_id: requestId }), }); @@ -2053,7 +2077,7 @@ export function subscribeSessionEvents( const headers: Record = { Accept: 'text/event-stream' }; if (lastEventId) headers['Last-Event-ID'] = lastEventId; - const res = await fetch(`${API}/sessions/${sessionId}/events`, { + const res = await fetch(`${API}${sessionPath(sessionId)}/events`, { headers, signal: controller.signal, }); diff --git a/src/components/MindSpacePageDetail.tsx b/src/components/MindSpacePageDetail.tsx index c9dcacf..4eb3772 100644 --- a/src/components/MindSpacePageDetail.tsx +++ b/src/components/MindSpacePageDetail.tsx @@ -13,6 +13,7 @@ import { publishPageToPlaza, redactMindSpacePage, updateMindSpacePage, + updatePublicationStatus, } from '../api/client'; import type { MindSpacePage, @@ -288,6 +289,8 @@ export function MindSpacePageDetail({ pageTitle: string; pushedToPlaza: boolean; } | null>(null); + const [confirmPublicationStatusOpen, setConfirmPublicationStatusOpen] = useState(false); + const [statusConfirming, setStatusConfirming] = useState(false); const applyPageRecord = useCallback( (next: MindSpacePage, options?: { recordHistory?: boolean }) => { @@ -334,6 +337,13 @@ export function MindSpacePageDetail({ ? await getMindSpacePublicationStats(next.publication.id).catch(() => null) : null, ); + if ( + next.publication && + next.publication.accessMode === 'public' && + next.publication.userConfirmedAt === null + ) { + setConfirmPublicationStatusOpen(true); + } } catch (err) { setError(err instanceof Error ? err.message : '页面加载失败'); } finally { @@ -434,6 +444,34 @@ export function MindSpacePageDetail({ } }; + const confirmPublicationStatus = async (nextAccessMode: AccessMode) => { + if (!page || !page.publication) return; + setStatusConfirming(true); + try { + const nextExpiresAt = + nextAccessMode === 'public' + ? null + : nextAccessMode === 'time_limited' + ? Date.now() + 30 * 60 * 1000 + : null; + const updated = await updatePublicationStatus(page.publication.id, { + accessMode: nextAccessMode, + expiresAt: nextExpiresAt, + }); + setPage({ ...page, publication: updated }); + setAccessMode(updated.accessMode); + setExpiresAt( + updated.expiresAt ? localDateTimeValue(updated.expiresAt) : '', + ); + setConfirmPublicationStatusOpen(false); + setError(null); + } catch (err) { + setError(err instanceof Error ? err.message : '更新发布状态失败'); + } finally { + setStatusConfirming(false); + } + }; + const offlineFromSuccess = async () => { if (!publishSuccess) return; setPublishing(true); @@ -1395,7 +1433,13 @@ export function MindSpacePageDetail({ ))} - {sharePayload ? setSharePayload(null)} /> : null} + {sharePayload ? ( + setSharePayload(null)} + /> + ) : null} {fullscreenPreviewOpen && page && page.contentFormat === 'html' ? ( ) : null} + + {confirmPublicationStatusOpen && page?.publication ? ( + +
+

页面预览期确认

+

你的页面当前处于 30 分钟预览期。请选择后续状态：

+
+ + + +
+
+ + ) : null} ); } diff --git a/src/components/ShareSheet.tsx b/src/components/ShareSheet.tsx index d97c7dd..eb00fc8 100644 --- a/src/components/ShareSheet.tsx +++ b/src/components/ShareSheet.tsx @@ -1,4 +1,5 @@ -import { useState } from 'react'; +import { useState, useEffect } from 'react'; +import type { MindSpacePublication } from '../types'; import { SHARE_CHANNELS, canUseNativeShare, @@ -10,13 +11,37 @@ import { export function ShareSheet({ payload, + publication, onClose, }: { payload: SharePayload; + publication?: MindSpacePublication | null; onClose: () => void; }) { const [message, setMessage] = useState(null); const [busy, setBusy] = useState(false); + const [expiresInMinutes, setExpiresInMinutes] = useState(null); + + useEffect(() => { + if (!publication || publication.accessMode !== 'public' || publication.userConfirmedAt !== null) { + setExpiresInMinutes(null); + return; + } + if (!publication.expiresAt) { + setExpiresInMinutes(null); + return; + } + + const update = () => { + const now = Date.now(); + const remaining = Math.ceil((publication.expiresAt! - now) / (1000 * 60)); + setExpiresInMinutes(Math.max(0, remaining)); + }; + update(); + + const interval = setInterval(update, 10000); + return () => clearInterval(interval); + }, [publication]); const handleChannel = async (channel: ShareChannel) => { setBusy(true); @@ -60,6 +85,15 @@ export function ShareSheet({

SHARE

分享到

{payload.title}

+ {expiresInMinutes !== null && ( +

+ ⏱️ 预览还剩{' '} + + {expiresInMinutes > 0 ? `${expiresInMinutes} 分钟` : '不到 1 分钟'} + + ，之后仅你能看到。确认发布为永久公开？ +

+ )}