backend/app/services/desktop_tasks.py

@@ -0,0 +1,106 @@
+"""Desktop Task records are committed by Host before returning to Core callers."""
+from __future__ import annotations
+from datetime import datetime, timezone
+import re
+from uuid import uuid4, uuid5, NAMESPACE_URL
+from app import host_bridge
+from app.contracts import Task, TaskStatus
+from app.database.db import connect_knowledge, transaction
+from app.errors import ApiError
+from app.services import desktop_notes
+
+
+def _call(method, **params): return desktop_notes.call('records.' + method, **params)
+def _ms(value): return None if value is None else int(value.timestamp() * 1000)
+def _datetime(value): return None if value is None else datetime.fromtimestamp(value / 1000, timezone.utc)
+def _record(task):
+    return {'schema': 1, 'kind': 'task', 'id': task.task_id, 'data': {
+        'title': task.title, 'description': task.description, 'status': task.status.value,
+        'note_id': task.note_id, 'due_at_ms': _ms(task.due_at),
+        'created_at_ms': _ms(task.created_at), 'updated_at_ms': _ms(task.updated_at)}}
+def _task(record):
+    data = record['data']
+    return Task(task_id=record['id'], title=data['title'], description=data['description'], status=data['status'],
+                note_id=data['note_id'], due_at=_datetime(data['due_at_ms']), created_at=_datetime(data['created_at_ms']), updated_at=_datetime(data['updated_at_ms']))
+def _operation(): return host_bridge.operation_id.get() or str(uuid4())
+def _replay(operation, task_id=None, values=None, deleted=False):
+    previous = _call('operation', operation_id=operation)
+    if previous is None: return None
+    if previous.get('state') != 'committed' or previous.get('deleted') != deleted:
+        raise ApiError(409, 'OPERATION_PAYLOAD_CONFLICT', '该操作标识已用于其他修改。')
+    task = _task(previous['record'])
+    if task_id is not None and task.task_id != task_id:
+        raise ApiError(409, 'OPERATION_PAYLOAD_CONFLICT', '该操作标识已用于其他任务。')
+    for name, value in (values or {}).items():
+        actual = getattr(task, name)
+        if isinstance(actual, datetime) and isinstance(value, datetime):
+            actual, value = _ms(actual), _ms(value)
+        if actual != value: raise ApiError(409, 'OPERATION_PAYLOAD_CONFLICT', '该操作标识的字段不一致。')
+    return task
+
+def _migrate():
+    # Only the already scoped Vault database is eligible; unassigned legacy global data stays untouched.
+    conn = connect_knowledge()
+    try:
+        if conn.execute("SELECT value FROM index_meta WHERE key='tasks_host_owned_v1'").fetchone(): return
+        from app.services.task_service import _task_from_row
+        for row in conn.execute('SELECT * FROM tasks ORDER BY task_id').fetchall():
+            task = _task_from_row(row)
+            if _call('get', id=task.task_id) is None:
+                operation = str(uuid5(NAMESPACE_URL, 'opennexus-task-migration:' + host_bridge.vault_id.get() + ':' + task.task_id))
+                _call('write', record=_record(task), expected='', operation_id=operation)
+        with transaction(conn):
+            conn.execute("INSERT OR REPLACE INTO index_meta VALUES ('tasks_host_owned_v1','1')")
+    finally: conn.close()
+
+def _link(note_id):
+    if not note_id: return None
+    try: return desktop_notes.call('read', file_id=note_id)['file_id']
+    except ApiError as error:
+        if error.code == 'FILE_NOT_FOUND': raise ApiError(404, 'RESOURCE_NOT_FOUND', 'note not found', {'note_id': note_id}) from None
+        raise
+
+def create(*, title, description='', note_id=None, due_at=None):
+    _migrate(); operation = _operation()
+    values = {'title': title, 'description': description, 'note_id': note_id, 'due_at': due_at}
+    replay = _replay(operation, values=values)
+    if replay is not None: return replay
+    now = datetime.now(timezone.utc)
+    task_id = 'task_' + uuid5(NAMESPACE_URL, 'opennexus-task:' + operation).hex
+    task = Task(task_id=task_id, title=title, description=description, note_id=_link(note_id), due_at=due_at, created_at=now, updated_at=now)
+    receipt = _call('write', record=_record(task), expected='', operation_id=operation)
+    return _task(receipt['record'])
+def get(task_id):
+    if re.fullmatch(r'task_[0-9a-f]{32}', task_id) is None: return None
+    _migrate(); value = _call('get', id=task_id)
+    return _task(value['record']) if value is not None else None
+def list_tasks(*, limit, offset):
+    _migrate(); result = _call('list', limit=1000, offset=0); records = list(result['items'])
+    while len(records) < result['total']:
+        page = _call('list', limit=1000, offset=len(records))
+        if not page['items']: break
+        records.extend(page['items'])
+    tasks = sorted((_task(value['record']) for value in records), key=lambda value: (value.updated_at, value.task_id), reverse=True)
+    return tasks[offset:offset+limit], len(tasks)
+def update(task_id, values):
+    _migrate(); operation = _operation(); values = dict(values)
+    for key in ['title', 'description', 'status']:
+        if values.get(key) is None: values.pop(key, None)
+    if not set(values) <= {'title','description','status','note_id','due_at'}: raise ApiError(422, 'INVALID_ARGUMENT', '未知任务字段。')
+    replay = _replay(operation, task_id, values)
+    if replay is not None: return replay
+    current = _call('get', id=task_id)
+    if current is None: raise ApiError(404, 'RESOURCE_NOT_FOUND', 'task not found', {'task_id': task_id})
+    if 'note_id' in values: values['note_id'] = _link(values['note_id'])
+    task = _task(current['record']).model_copy(update={**values, 'updated_at': datetime.now(timezone.utc)})
+    if isinstance(task.status, str): task.status = TaskStatus(task.status)
+    receipt = _call('write', record=_record(task), expected=current['hash'], operation_id=operation)
+    return _task(receipt['record'])
+def delete(task_id):
+    if re.fullmatch(r'task_[0-9a-f]{32}', task_id) is None: return False
+    _migrate(); operation = _operation()
+    if _replay(operation, task_id, deleted=True) is not None: return True
+    current = _call('get', id=task_id)
+    if current is None: return False
+    _call('delete', id=task_id, expected=current['hash'], operation_id=operation)
+    return True

backend/app/services/task_service.py

@@ -13,6 +13,11 @@ from app.database.db import connect_knowledge as connect, transaction
 from app.errors import ApiError
 from app.operation_logs import log_event
 
+def _desktop():
+    from app.config import get_settings
+    return get_settings().environment == 'desktop'
+
+
 _write_locks = WeakKeyDictionary()
 
 
@@ -63,6 +68,9 @@ def create_task(
     *, title: str, description: str = "", note_id: str | None = None,
     due_at: datetime | None = None,
 ) -> Task:
+    if _desktop():
+        from app.services import desktop_tasks
+        return desktop_tasks.create(title=title, description=description, note_id=note_id, due_at=due_at)
     _prepare_note_link(note_id)
     if note_id and repository.get_note_record(note_id) is None:
         raise ApiError(404, "RESOURCE_NOT_FOUND", "note not found", {"note_id": note_id})
@@ -90,6 +98,9 @@ def create_task(
 
 
 def get_task(task_id: str) -> Task | None:
+    if _desktop():
+        from app.services import desktop_tasks
+        return desktop_tasks.get(task_id)
     conn = connect()
     try:
         row = conn.execute("SELECT * FROM tasks WHERE task_id = ?", (task_id,)).fetchone()
@@ -99,6 +110,9 @@ def get_task(task_id: str) -> Task | None:
 
 
 def list_tasks(*, limit: int, offset: int) -> tuple[list[Task], int]:
+    if _desktop():
+        from app.services import desktop_tasks
+        return desktop_tasks.list_tasks(limit=limit, offset=offset)
     conn = connect()
     try:
         total = conn.execute("SELECT COUNT(*) FROM tasks").fetchone()[0]
@@ -112,6 +126,9 @@ def list_tasks(*, limit: int, offset: int) -> tuple[list[Task], int]:
 
 
 def update_task(task_id: str, values: dict[str, object]) -> Task:
+    if _desktop():
+        from app.services import desktop_tasks
+        return desktop_tasks.update(task_id, values)
     current = get_task(task_id)
     if current is None:
         raise ApiError(404, "RESOURCE_NOT_FOUND", "task not found", {"task_id": task_id})
@@ -154,6 +171,9 @@ def update_task(task_id: str, values: dict[str, object]) -> Task:
 
 
 def delete_task(task_id: str) -> bool:
+    if _desktop():
+        from app.services import desktop_tasks
+        return desktop_tasks.delete(task_id)
     conn = connect()
     try:
         with transaction(conn):

backend/tests/test_desktop_projection.py

@@ -40,18 +40,19 @@ def test_projection_isolates_same_path_and_refreshes_changed_deleted_content(tmp
     assert 'file-a' not in search(first, 'replacementtoken').model_dump_json()
     assert (tmp_path/'vault-state'/first/'core.sqlite3').is_file()
     assert (tmp_path/'vault-state'/second/'core.sqlite3').is_file()
-    from app.services import task_service
     token = host_bridge.vault_id.set(second)
     try:
-        task = task_service.create_task(title='Scoped task', note_id='file-b')
-        assert task.note_id == 'file-b'
-    finally:
-        host_bridge.vault_id.reset(token)
+        conn = db.connect_knowledge()
+        with db.transaction(conn):
+            conn.execute("INSERT INTO tasks VALUES ('legacy-task','Scoped task','','todo','file-b',NULL,'2026-09-08T00:00:00+00:00','2026-09-08T00:00:00+00:00')")
+        conn.close()
+    finally: host_bridge.vault_id.reset(token)
     token = host_bridge.vault_id.set(first)
     try:
-        assert task_service.get_task(task.task_id) is None
-    finally:
-        host_bridge.vault_id.reset(token)
+        conn = db.connect_knowledge()
+        assert conn.execute("SELECT * FROM tasks WHERE task_id='legacy-task'").fetchone() is None
+        conn.close()
+    finally: host_bridge.vault_id.reset(token)
 
 
 def test_desktop_semantic_rebuild_preserves_host_file_id(tmp_path, monkeypatch):
@@ -82,14 +83,19 @@ def test_host_identity_adoption_keeps_existing_task_links(tmp_path, monkeypatch)
     monkeypatch.setattr('app.config.get_settings', lambda: settings)
     document = {'file_id': 'before-merge', 'path': 'same.md', 'hash': sha256(b'test').hexdigest(), 'content': 'test', 'created_at': 0, 'updated_at': 1}
     monkeypatch.setattr(desktop_notes, 'call', lambda method, **params: {'items': [document], 'total': 1} if method == 'list' else document)
-    from app.services import task_service
     token = host_bridge.vault_id.set(str(uuid4()))
     try:
-        task = task_service.create_task(title='Preserve link', note_id='before-merge')
+        asyncio.run(desktop_projection.refresh())
+        conn = db.connect_knowledge()
+        with db.transaction(conn):
+            conn.execute("INSERT INTO tasks VALUES ('legacy-task','Preserve link','','todo','before-merge',NULL,'2026-09-08T00:00:00+00:00','2026-09-08T00:00:00+00:00')")
+        conn.close()
         document['file_id'] = 'after-merge'
         document['aliases'] = ['before-merge']
         asyncio.run(desktop_projection.refresh())
-        assert task_service.get_task(task.task_id).note_id == 'after-merge'
+        conn = db.connect_knowledge()
+        assert conn.execute("SELECT note_id FROM tasks WHERE task_id='legacy-task'").fetchone()['note_id'] == 'after-merge'
+        conn.close()
         assert repository.get_note_record('before-merge') is None
         assert repository.get_note_record('after-merge') is not None
     finally:

docs/contracts/Sync-v1契约.md

@@ -53,3 +53,14 @@ Vault 行锁内执行幂等键验证、CAS、路径检查、序列分配及历
 当前历史和 tombstone **永久保留**，游标不主动过期；不启用 GC，以免缺少恢复演练时删除历史对象。配额包含全部历史对象，未完成上传预留额度。通知尚未实现，客户端必须主动按 cursor 拉取。上传过期清理、对象 GC、限速指标、完整备份恢复工具与生产负载验收仍未交付。`/ready` 目前只检查数据库 schema，不证明 S3 就绪。
 
 验证向量见 `server sync/tests/test_protocol.py`：双设备 CAS、幂等重放、移动与删除、历史恢复、稳定分页、隔离、撤销、刷新、配额、路径冲突、摘要与断点。
+
+
+## 逻辑任务记录 v1
+
+普通文件传输增加保留命名空间 `opennexus-records/v1/tasks/task_<32位小写十六进制>.json`。文件 Revision 的 file_id 与任务业务 id 分开；记录仅包含 `schema=1`、`kind=task`、`id`、`data`。data 白名单为 title、description、status、note_id、due_at_ms、created_at_ms、updated_at_ms。时间采用 UTC Unix 毫秒，due_at_ms/note_id 可为空；status 为 todo/in_progress/done/cancelled。任务状态只表示任务状态，不启动目标设备上的 Agent 或后台作业。
+
+Host 在写入 journal、捕获外部修改和上传前验证记录；未知字段、未知 schema/kind、非法路径或时间拒绝。记录不允许 api_key、token、environment、permissions 等附加字段。标题最多 4096 字节、说明最多 256 KiB、记录最多 1 MiB，Core list 响应最多 4 MiB。用户写入标题/说明的正文仍是用户内容，不按关键字审查正文。
+
+桌面 Task CRUD 经 Host records broker，操作重放返回原始记录且同 ID 不同字段拒绝；Core 不以全局 SQLite 作为任务来源。已明确归属于当前 Vault 的旧 Task 表在首次访问时逐条迁移，完成后设置所有权标记，来源表保留；未分配 Vault 的全局旧数据不猜测归属。文件身份采纳时，任务链接通过 Host 别名解析，并在上传队列物化前将规范化引用写成新的逻辑记录。
+
+本节仅交付任务逻辑记录。用户 Skill/配置、主题、可选对话等仍须逐类定义白名单与适配器；不得用复制任意 JSON/SQLite 代替。

docs/development/OpenNexus生产化实施进度-2026-09-08.md

@@ -90,3 +90,13 @@ Core 的独立数据目录目前不等于已授权 Vault。Python 旧笔记写
 - 目前快照最多处理 100000 条历史 revision，超限明确拒绝；大库快照/扫描性能、任务和配置逻辑同步、完整分类故障矩阵及签名/沙箱等其他生产门槛继续实施。
 
 - 本增量后端全量 900 项、Rust desktop 全目标 49 项通过（另 1 个父测试使用的进程辅助入口）；Sync 界面 3 项交互测试、两个 TypeScript 项目检查和 Clippy `-D warnings` 通过。
+
+
+## 增量：Task 逻辑记录
+
+- Task v1 使用专用命名空间和 Rust 字段白名单，秘密字段、未知字段/版本在 journal 前拒绝；普通配置文件仍不纳入同步。任务不依赖 Core SQLite 的全量复制，也不触发目标设备自动执行任务。
+- 桌面 Task CRUD 已交由 Host 持久文件和 outbox；同操作创建重复 20 次返回同一结果，不同字段重用 ID 拒绝；更新/删除同样通过操作回执。已有当前 Vault 任务逐条迁移，保留原表且可重复完成迁移。
+- 真实 Core HTTP 验证任务创建/更新/删除/重放、跨 Vault 拒绝、旧任务来源保留；真实 Sync HTTP 验证两客户端任务状态与删除传播。记录列表有 4 MiB 分页上限，读取有 1 MiB 文件上限。
+- 本地文件 ID 采纳后的 task note_id 通过别名解析，上传前生成持久规范化引用，避免把本机旧 ID 留给另一设备。其余逻辑记录类别继续实施。
+
+- Task 增量后端全量 900 项、Rust desktop 全目标 50 项通过（另 1 个父测试驱动的进程辅助入口），Clippy `-D warnings` 通过。

frontend/src-tauri/src/lib.rs

@@ -4,6 +4,7 @@ pub mod core;
 pub mod credentials;
 mod payloads;
 pub mod recent;
+pub mod records;
 #[cfg(feature = "desktop")]
 pub mod request_lifecycle;
 mod runtime_compat;

frontend/src-tauri/src/records.rs

@@ -0,0 +1,217 @@
+//! Versioned logical records: explicit fields only, never raw application databases/config.
+use crate::workspace::{HostError, Result, Workspace};
+use serde::{Deserialize, Serialize};
+use serde_json::{json, Value};
+#[derive(Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct TaskData {
+    pub title: String,
+    pub description: String,
+    pub status: String,
+    pub note_id: Option<String>,
+    pub due_at_ms: Option<i64>,
+    pub created_at_ms: i64,
+    pub updated_at_ms: i64,
+}
+#[derive(Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct Record {
+    pub schema: u32,
+    pub kind: String,
+    pub id: String,
+    pub data: TaskData,
+}
+pub fn path(id: &str) -> Result<String> {
+    if !id.starts_with("task_")
+        || id.len() != 37
+        || !id[5..]
+            .bytes()
+            .all(|v| v.is_ascii_digit() || (b'a'..=b'f').contains(&v))
+    {
+        return Err(HostError::new("RECORD_ID_INVALID"));
+    }
+    Ok(format!("opennexus-records/v1/tasks/{id}.json"))
+}
+pub fn is_record(path: &str) -> bool {
+    path.starts_with("opennexus-records/")
+}
+pub fn allowed(path_value: &str) -> bool {
+    path_value
+        .strip_prefix("opennexus-records/v1/tasks/")
+        .and_then(|v| v.strip_suffix(".json"))
+        .is_some_and(|id| path(id).is_ok())
+}
+pub fn validate(path_value: &str, content: &[u8]) -> Result<Record> {
+    if content.len() > 1024 * 1024 {
+        return Err(HostError::new("RECORD_TOO_LARGE"));
+    }
+    let record: Record =
+        serde_json::from_slice(content).map_err(|_| HostError::new("RECORD_SCHEMA_INVALID"))?;
+    let time = |value: i64| (0..=253402300799999).contains(&value);
+    if record.schema != 1 || record.kind != "task" || path(&record.id)? != path_value {
+        return Err(HostError::new("RECORD_SCHEMA_UNSUPPORTED"));
+    }
+    let data = &record.data;
+    if data.title.trim().is_empty()
+        || data.title.len() > 4096
+        || data.description.len() > 262144
+        || !matches!(
+            data.status.as_str(),
+            "todo" | "in_progress" | "done" | "cancelled"
+        )
+        || !time(data.created_at_ms)
+        || !time(data.updated_at_ms)
+        || data.due_at_ms.is_some_and(|v| !time(v))
+        || data.note_id.as_ref().is_some_and(|v| {
+            v.is_empty()
+                || v.len() > 128
+                || !v
+                    .bytes()
+                    .all(|b| b.is_ascii_alphanumeric() || b"_-".contains(&b))
+        })
+    {
+        return Err(HostError::new("RECORD_DATA_INVALID"));
+    }
+    Ok(record)
+}
+impl Workspace {
+    pub(crate) fn normalize_record_links(&mut self) -> Result<()> {
+        for path in self.sync_paths()?.into_iter().filter(|v| allowed(v)) {
+            let bytes = std::fs::read(self.resolve(&path)?)?;
+            let original = validate(&path, &bytes)?;
+            if let Some(note_id) = original.data.note_id.as_ref() {
+                if let Ok(note_path) = self.path_for_id(note_id) {
+                    if let Some(entry) = self.entry(&note_path)? {
+                        if &entry.file_id != note_id {
+                            let mut record = original;
+                            record.data.note_id = Some(entry.file_id);
+                            self.write(
+                                &path,
+                                &crate::workspace::hash(&bytes),
+                                &serde_json::to_vec(&record)
+                                    .map_err(|_| HostError::new("RECORD_SCHEMA_INVALID"))?,
+                                "local",
+                            )?;
+                        }
+                    }
+                }
+            }
+        }
+        Ok(())
+    }
+    pub fn record_get(&mut self, id: &str) -> Result<Option<Value>> {
+        let path = path(id)?;
+        if !self.resolve(&path)?.is_file() {
+            return Ok(None);
+        }
+        if std::fs::metadata(self.resolve(&path)?)?.len() > 1024 * 1024 {
+            return Err(HostError::new("RECORD_TOO_LARGE"));
+        }
+        let document = self.read(&path)?;
+        let mut record = validate(&path, document.content.as_bytes())?;
+        if let Some(note_id) = record.data.note_id.as_ref() {
+            if let Ok(path) = self.path_for_id(note_id) {
+                if let Some(entry) = self.entry(&path)? {
+                    record.data.note_id = Some(entry.file_id);
+                }
+            }
+        }
+        Ok(Some(
+            json!({"record":record,"hash":document.entry.hash,"file_id":document.entry.file_id}),
+        ))
+    }
+    pub fn record_list(&mut self, offset: usize, limit: usize) -> Result<Value> {
+        if limit == 0 || limit > 1000 {
+            return Err(HostError::new("RECORD_LIMIT_INVALID"));
+        }
+        let paths = self
+            .sync_paths()?
+            .into_iter()
+            .filter(|path| allowed(path))
+            .collect::<Vec<_>>();
+        let total = paths.len();
+        let mut items = Vec::new();
+        let mut bytes = 0;
+        for path in paths.into_iter().skip(offset).take(limit) {
+            let id = path
+                .strip_prefix("opennexus-records/v1/tasks/")
+                .and_then(|v| v.strip_suffix(".json"))
+                .ok_or_else(|| HostError::new("RECORD_ID_INVALID"))?;
+            if let Some(value) = self.record_get(id)? {
+                let size = serde_json::to_vec(&value)
+                    .map_err(|_| HostError::new("RECORD_SCHEMA_INVALID"))?
+                    .len();
+                if bytes + size > 4 * 1024 * 1024 {
+                    break;
+                }
+                bytes += size;
+                items.push(value);
+            }
+        }
+        Ok(json!({"items":items,"total":total}))
+    }
+    pub fn record_operation(&mut self, operation: &str) -> Result<Option<Value>> {
+        self.recover()?;
+        let Some(receipt) = self.operation(operation)? else {
+            return Ok(None);
+        };
+        let path = receipt["result"]["path"]
+            .as_str()
+            .ok_or_else(|| HostError::new("RECORD_OPERATION_PENDING"))?;
+        if !allowed(path) {
+            return Err(HostError::new("RECORD_OPERATION_DENIED"));
+        }
+        let bytes = self.payload(operation, &[])?;
+        let record = validate(path, &bytes)?;
+        Ok(Some(
+            json!({"record":record,"deleted":receipt["result"]["deleted"],"state":receipt["state"]}),
+        ))
+    }
+}
+#[cfg(test)]
+mod tests {
+    use super::*;
+    fn fixture() -> Value {
+        json!({"schema":1,"kind":"task","id":"task_00000000000000000000000000000001","data":{"title":"Task","description":"","status":"todo","note_id":null,"due_at_ms":null,"created_at_ms":0,"updated_at_ms":0}})
+    }
+    #[test]
+    fn whitelist_rejects_secret_unknown_fields_and_future_schema_before_journal() {
+        let root = tempfile::tempdir().unwrap();
+        let mut ws = Workspace::open(root.path()).unwrap();
+        let value = fixture();
+        let path = path(value["id"].as_str().unwrap()).unwrap();
+        for field in ["api_key", "token", "environment", "permissions", "extra"] {
+            let mut value = value.clone();
+            value["data"][field] = json!("planted-secret");
+            assert_eq!(
+                ws.write(&path, "", &serde_json::to_vec(&value).unwrap(), "local")
+                    .unwrap_err()
+                    .code,
+                "RECORD_SCHEMA_INVALID"
+            );
+        }
+        let mut future = value.clone();
+        future["schema"] = json!(2);
+        assert_eq!(
+            ws.write(&path, "", &serde_json::to_vec(&future).unwrap(), "remote")
+                .unwrap_err()
+                .code,
+            "RECORD_SCHEMA_UNSUPPORTED"
+        );
+        assert!(!root.path().join(&path).exists());
+        assert_eq!(ws.pending_count().unwrap(), 0);
+        let operation = uuid::Uuid::new_v4().to_string();
+        ws.write_operation(
+            &path,
+            "",
+            &serde_json::to_vec(&value).unwrap(),
+            "local",
+            &operation,
+        )
+        .unwrap();
+        assert_eq!(
+            ws.record_operation(&operation).unwrap().unwrap()["record"],
+            value
+        );
+    }
+}

frontend/src-tauri/src/sync_discovery.rs

@@ -5,6 +5,9 @@ use std::{collections::HashSet, fs, io::Read, path::Path};
 use uuid::Uuid;
 /// File transport only. Logical records receive their own versioned whitelist separately.
 pub fn allowed(path: &str) -> bool {
+    if crate::records::is_record(path) {
+        return crate::records::allowed(path);
+    }
     let parts: Vec<_> = path.split('/').collect();
     if parts.iter().any(|part| {
         part.starts_with('.')
@@ -100,6 +103,9 @@ impl Workspace {
             if bytes.len() > 104857600 {
                 return Err(HostError::new("FILE_TOO_LARGE"));
             }
+            if crate::records::is_record(&path) {
+                crate::records::validate(&path, &bytes)?;
+            }
             let digest = hash(&bytes);
             let previous = self.entry(&path)?;
             let observed: Option<(String, String, bool)> = if let Some(entry) = &previous {

frontend/src-tauri/src/sync_state.rs

@@ -140,6 +140,7 @@ impl Workspace {
     }
     pub fn sync_capture(&mut self, binding: &str) -> Result<()> {
         self.check_binding(binding)?;
+        self.normalize_record_links()?;
         loop {
             let pending = self.db.query_row("SELECT operation_id,file_id,path,hash,operation,content FROM outbox WHERE state='pending' ORDER BY rowid LIMIT 1", [], |r| {
                 Ok((r.get::<_,String>(0)?,r.get::<_,String>(1)?,r.get::<_,String>(2)?,r.get::<_,String>(3)?,r.get::<_,String>(4)?,r.get::<_,Vec<u8>>(5)?))
@@ -165,6 +166,9 @@ impl Workspace {
                     return Err(HostError::new("SYNC_SPOOL_CORRUPT"));
                 }
                 crate::payloads::verify(&self.sync_spool(&digest)?, &digest, size as u64)?;
+                if crate::records::is_record(&path) {
+                    crate::records::validate(&path, &self.payload(&operation_id, &content)?)?;
+                }
                 size
             } else {
                 0

frontend/src-tauri/src/workspace.rs

@@ -258,10 +258,11 @@ impl Workspace {
                 .map_err(|_| HostError::new("UNSAFE_PATH"))?
                 .to_string_lossy()
                 .replace('\\', "/");
-            if relative
-                .split('/')
-                .any(|s| s.eq_ignore_ascii_case(".ainote") || s.eq_ignore_ascii_case(".git"))
-                || linked(&path)?
+            if relative.split('/').any(|s| {
+                s.eq_ignore_ascii_case(".ainote")
+                    || s.eq_ignore_ascii_case(".git")
+                    || s.eq_ignore_ascii_case("opennexus-records")
+            }) || linked(&path)?
             {
                 continue;
             }
@@ -407,6 +408,9 @@ impl Workspace {
         if Uuid::parse_str(operation_id).is_err() {
             return Err(HostError::new("OPERATION_ID_INVALID"));
         }
+        if crate::records::is_record(path) {
+            crate::records::validate(path, content)?;
+        }
         if content.len() > 100 * 1024 * 1024 {
             return Err(HostError::new("FILE_TOO_LARGE"));
         }
@@ -526,6 +530,9 @@ impl Workspace {
         content: &[u8],
         origin: &str,
     ) -> Result<()> {
+        if crate::records::is_record(path) {
+            crate::records::validate(path, content)?;
+        }
         let target = self.resolve(path)?;
         let digest = hash(content);
         let current = if target.exists() {

frontend/src-tauri/src/workspace_broker.rs

@@ -42,6 +42,28 @@ struct Mutation {
     kind: String,
 }
 
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+struct RecordRead {
+    vault_id: String,
+    id: String,
+}
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+struct RecordDelete {
+    vault_id: String,
+    id: String,
+    expected: String,
+    operation_id: String,
+}
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+struct RecordWrite {
+    vault_id: String,
+    record: Value,
+    expected: String,
+    operation_id: String,
+}
 fn bound(ws: &Workspace, vault_id: &str) -> Result<(), String> {
     if ws.vault_id != vault_id {
         return Err("VAULT_PERMISSION_CHANGED".into());
@@ -54,6 +76,47 @@ fn decode<T: serde::de::DeserializeOwned>(value: &Value) -> Result<T, String> {
 pub fn dispatch(ws: &mut Workspace, request: &Value) -> Result<Value, String> {
     let params = &request["params"];
     match request["rpc"].as_str().unwrap_or_default() {
+        "workspace.records.list" => {
+            let p: List = decode(params)?;
+            bound(ws, &p.vault_id)?;
+            ws.record_list(p.offset, p.limit).map_err(|e| e.code)
+        }
+        "workspace.records.get" => {
+            let p: RecordRead = decode(params)?;
+            bound(ws, &p.vault_id)?;
+            ws.record_get(&p.id)
+                .map(|v| v.unwrap_or(Value::Null))
+                .map_err(|e| e.code)
+        }
+        "workspace.records.operation" => {
+            let p: Operation = decode(params)?;
+            bound(ws, &p.vault_id)?;
+            ws.record_operation(&p.operation_id)
+                .map(|v| v.unwrap_or(Value::Null))
+                .map_err(|e| e.code)
+        }
+        "workspace.records.delete" => {
+            let p: RecordDelete = decode(params)?;
+            bound(ws, &p.vault_id)?;
+            let path = crate::records::path(&p.id).map_err(|e| e.code)?;
+            ws.mutate_operation("delete", &path, "", &p.expected, &p.operation_id)
+                .map_err(|e| e.code)?;
+            ws.record_operation(&p.operation_id)
+                .map(|v| v.unwrap_or(Value::Null))
+                .map_err(|e| e.code)
+        }
+        "workspace.records.write" => {
+            let p: RecordWrite = decode(params)?;
+            bound(ws, &p.vault_id)?;
+            let path =
+                crate::records::path(p.record["id"].as_str().unwrap_or("")).map_err(|e| e.code)?;
+            let bytes = serde_json::to_vec(&p.record).map_err(|_| "RECORD_SCHEMA_INVALID")?;
+            ws.write_operation(&path, &p.expected, &bytes, "local", &p.operation_id)
+                .map_err(|e| e.code)?;
+            ws.record_operation(&p.operation_id)
+                .map(|v| v.unwrap_or(Value::Null))
+                .map_err(|e| e.code)
+        }
         "workspace.list" => {
             let p: List = decode(params)?;
             bound(ws, &p.vault_id)?;

frontend/src-tauri/tests/core_workspace.rs

@@ -146,6 +146,122 @@ async fn real_core_notes_roundtrip_only_through_bound_host_and_confirm_commits()
     .await;
     assert_eq!(status, 409, "{denied}");
     assert_eq!(denied["error"]["code"], "VAULT_PERMISSION_CHANGED");
+
+    let legacy_database = rusqlite::Connection::open(
+        temp.path()
+            .join("core/vault-state")
+            .join(&vault)
+            .join("core.sqlite3"),
+    )
+    .unwrap();
+    legacy_database.execute("INSERT INTO tasks VALUES ('task_00000000000000000000000000000002','Legacy task','','todo',?1,NULL,'2026-09-08T00:00:00+00:00','2026-09-08T00:00:00+00:00')",[file_id]).unwrap();
+    let task_operation = uuid::Uuid::new_v4().to_string();
+    let task_body = json!({"title":"Host task","note_id":file_id});
+    let (status, task) = request(
+        &mut core,
+        "POST",
+        "/api/tasks",
+        &vault,
+        &task_operation,
+        Some(task_body.clone()),
+    )
+    .await;
+    assert_eq!(status, 200, "{task}");
+    let task_id = task["task_id"].as_str().unwrap();
+    for _ in 0..20 {
+        let (status, replay) = request(
+            &mut core,
+            "POST",
+            "/api/tasks",
+            &vault,
+            &task_operation,
+            Some(task_body.clone()),
+        )
+        .await;
+        assert_eq!(status, 200, "{replay}");
+        assert_eq!(replay, task);
+    }
+    let (status, changed_request) = request(
+        &mut core,
+        "POST",
+        "/api/tasks",
+        &vault,
+        &task_operation,
+        Some(json!({"title":"changed","note_id":file_id})),
+    )
+    .await;
+    assert_eq!(status, 409, "{changed_request}");
+    let record_path = root.join(format!("opennexus-records/v1/tasks/{task_id}.json"));
+    assert!(record_path.is_file());
+    let (status, task_updated) = request(
+        &mut core,
+        "PATCH",
+        &format!("/api/tasks/{task_id}"),
+        &vault,
+        &uuid::Uuid::new_v4().to_string(),
+        Some(json!({"status":"done"})),
+    )
+    .await;
+    assert_eq!(status, 200, "{task_updated}");
+    assert_eq!(task_updated["status"], "done");
+    let (status, tasks) = request(
+        &mut core,
+        "GET",
+        "/api/tasks",
+        &vault,
+        &uuid::Uuid::new_v4().to_string(),
+        None,
+    )
+    .await;
+    assert_eq!(status, 200, "{tasks}");
+    assert_eq!(tasks["items"].as_array().unwrap().len(), 2);
+    let (status, task_denied) = request(
+        &mut core,
+        "GET",
+        "/api/tasks",
+        &uuid::Uuid::new_v4().to_string(),
+        &uuid::Uuid::new_v4().to_string(),
+        None,
+    )
+    .await;
+    assert_eq!(status, 409, "{task_denied}");
+    let deletion = uuid::Uuid::new_v4().to_string();
+    for _ in 0..2 {
+        let (status, result) = request(
+            &mut core,
+            "DELETE",
+            &format!("/api/tasks/{task_id}"),
+            &vault,
+            &deletion,
+            None,
+        )
+        .await;
+        assert_eq!(status, 200, "{result}");
+    }
+    assert!(!record_path.exists());
+    let source: String = legacy_database
+        .query_row(
+            "SELECT title FROM tasks WHERE task_id='task_00000000000000000000000000000002'",
+            [],
+            |r| r.get(0),
+        )
+        .unwrap();
+    assert_eq!(source, "Legacy task");
+    legacy_database
+        .execute("DELETE FROM index_meta WHERE key='tasks_host_owned_v1'", [])
+        .unwrap();
+    let (status, after_migration) = request(
+        &mut core,
+        "GET",
+        "/api/tasks",
+        &vault,
+        &uuid::Uuid::new_v4().to_string(),
+        None,
+    )
+    .await;
+    assert_eq!(status, 200, "{after_migration}");
+    assert_eq!(after_migration["items"].as_array().unwrap().len(), 1);
+
     let (status, deleted) = request(
         &mut core,
         "DELETE",
@@ -168,7 +284,7 @@ async fn real_core_notes_roundtrip_only_through_bound_host_and_confirm_commits()
     .await;
     assert_eq!(status, 200, "{search}");
     assert!(!search.to_string().contains(file_id));
-    assert_eq!(workspace.lock().unwrap().pending_count().unwrap(), 4);
+    assert_eq!(workspace.lock().unwrap().pending_count().unwrap(), 8);
     assert!(!temp
         .path()
         .join("core/unbound-vault/Core fixture.md")

frontend/src-tauri/tests/sync_push.rs

@@ -423,6 +423,67 @@ async fn actual_service_accepts_ordered_push_and_repeat_commit_without_duplicate
             .content,
         "only-local"
     );
+
+    let task_id = "task_00000000000000000000000000000001";
+    let task_path = format!("opennexus-records/v1/tasks/{task_id}.json");
+    let task_record = json!({"schema":1,"kind":"task","id":task_id,"data":{"title":"Synchronized task","description":"","status":"todo","note_id":first.file_id,"due_at_ms":null,"created_at_ms":0,"updated_at_ms":0}});
+    workspace
+        .lock()
+        .unwrap()
+        .write(
+            &task_path,
+            "",
+            &serde_json::to_vec(&task_record).unwrap(),
+            "local",
+        )
+        .unwrap();
+    client.push_one(&workspace, &binding).await.unwrap();
+    client_b.pull_page(&workspace_b, &binding_b).await.unwrap();
+    {
+        let mut ws = workspace_b.lock().unwrap();
+        let task = ws.record_get(task_id).unwrap().unwrap();
+        assert_eq!(task["record"], task_record);
+        let mut next = task["record"].clone();
+        next["data"]["status"] = json!("done");
+        ws.write(
+            &task_path,
+            task["hash"].as_str().unwrap(),
+            &serde_json::to_vec(&next).unwrap(),
+            "local",
+        )
+        .unwrap();
+    }
+    client_b.push_one(&workspace_b, &binding_b).await.unwrap();
+    client.pull_page(&workspace, &binding).await.unwrap();
+    assert_eq!(
+        workspace
+            .lock()
+            .unwrap()
+            .record_get(task_id)
+            .unwrap()
+            .unwrap()["record"]["data"]["status"],
+        "done"
+    );
+    {
+        let mut ws = workspace_b.lock().unwrap();
+        let task = ws.record_get(task_id).unwrap().unwrap();
+        ws.mutate_operation(
+            "delete",
+            &task_path,
+            "",
+            task["hash"].as_str().unwrap(),
+            &uuid::Uuid::new_v4().to_string(),
+        )
+        .unwrap();
+    }
+    client_b.push_one(&workspace_b, &binding_b).await.unwrap();
+    client.pull_page(&workspace, &binding).await.unwrap();
+    assert!(workspace
+        .lock()
+        .unwrap()
+        .record_get(task_id)
+        .unwrap()
+        .is_none());
     // Kill the actual client process after each durable 10 MiB server offset,
     // before its response reaches the client. The next process must query offset.
     use sha2::{Digest, Sha256};