diff --git a/docs/openapi/convey-clients.json b/docs/openapi/convey-clients.json index 07472583e..91ebd316c 100644 --- a/docs/openapi/convey-clients.json +++ b/docs/openapi/convey-clients.json @@ -63,6 +63,7 @@ "file_read_failed", "health_report_failed", "identity_busy", + "import_client_id_conflict", "import_conflict", "import_metadata_failed", "import_not_found", @@ -1304,9 +1305,160 @@ ] } }, + "/app/import/api/meta": { + "post": { + "description": "Update allowlisted metadata fields on a not-yet-started import.", + "operationId": "import.meta", + "requestBody": { + "content": { + "application/json": { + "example": { + "facet": "work", + "path": "/journal/imports/20260618_143022/source.m4a" + }, + "schema": { + "additionalProperties": true, + "properties": { + "client": { + "type": "object" + }, + "facet": { + "type": "string" + }, + "imported_via": { + "type": "string" + }, + "mime_type": { + "type": "string" + }, + "observer_handle": { + "type": "string" + }, + "original_filename": { + "type": "string" + }, + "path": { + "type": "string" + }, + "setting": { + "type": "string" + }, + "source_hint": { + "type": "string" + } + }, + "required": [ + "path" + ], + "type": "object" + } + } + }, + "required": true + }, + "responses": { + "200": { + "content": { + "application/json": { + "example": { + "path": "/journal/imports/20260618_143022/source.m4a", + "status": "ok", + "timestamp": "20260618_143022", + "updated": { + "facet": "work" + } + }, + "schema": { + "additionalProperties": true, + "properties": { + "path": { + "type": "string" + }, + "status": { + "type": "string" + }, + "timestamp": { + "type": "string" + }, + "updated": { + "type": "object" + } + }, + "required": [ + "status", + "path", + "timestamp", + "updated" + ], + "type": "object" + } + } + }, + "description": "Import metadata updated." + }, + "400": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/Error" + } + } + }, + "description": "The import path was missing or the import state is terminal.", + "x-reason-codes": [ + "invalid_operation_for_state", + "missing_required_field" + ] + }, + "403": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/Error" + } + } + }, + "description": "Access gate rejected a revoked paired-link identity.", + "x-reason-codes": [ + "pl_revoked" + ] + }, + "404": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/Error" + } + } + }, + "description": "Import metadata was not found.", + "x-reason-codes": [ + "import_not_found" + ] + }, + "500": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/Error" + } + } + }, + "description": "Import metadata could not be read or updated.", + "x-reason-codes": [ + "import_metadata_failed" + ] + } + }, + "summary": "Update import metadata", + "tags": [ + "import" + ] + } + }, "/app/import/api/save": { "post": { - "description": "Save an uploaded import file or pasted text into imports staging. Submit either file or text.", + "description": "Save an uploaded import file or pasted text into imports staging. Submit either file or text. client_item_id is required for idempotent native-client staging.", "operationId": "import.save", "requestBody": { "content": { @@ -1314,6 +1466,12 @@ "schema": { "additionalProperties": true, "properties": { + "client": { + "type": "object" + }, + "client_item_id": { + "type": "string" + }, "deterministic_only": { "type": "boolean" }, @@ -1333,78 +1491,556 @@ "setting": { "type": "string" }, + "source_hint": { + "type": "string" + }, "text": { "type": "string" } }, + "required": [ + "client_item_id" + ], "type": "object" } } }, - "description": "Multipart body with either file or text." + "description": "Multipart body with either file or text.", + "required": true }, "responses": { "200": { "content": { "application/json": { "example": { + "client_item_id": "ios-item-4f8b", + "diagnostics": { + "source_inference": "extension", + "timestamp_detection_method": "upload_fallback", + "timestamp_detection_model_called": false, + "timestamp_detection_no_match_reason": null + }, "facet": null, - "path": "/journal/imports/20260618_143022/source.txt", + "metadata": { + "client": {}, + "imported_via": "ios", + "mime_type": "audio/mp4", + "observer_handle": null, + "original_filename": "source.m4a", + "source_hint": null + }, + "path": "/journal/imports/20260618_143022/source.m4a", + "recommended_action": "start", + "replay": false, + "schema_version": 1, "setting": null, - "timestamp": "20260618_143022", - "timestamp_detection_method": "deterministic", - "timestamp_detection_model_called": false, - "timestamp_detection_no_match_reason": null + "source": "audio", + "status": "staged", + "timestamp": "20260618_143022" }, "schema": { "additionalProperties": true, "properties": { + "client_item_id": { + "type": "string" + }, + "diagnostics": { + "additionalProperties": true, + "properties": { + "source_inference": { + "enum": [ + "extension", + "content_type", + "default" + ], + "type": "string" + }, + "timestamp_detection_method": { + "type": "string" + }, + "timestamp_detection_model_called": { + "type": "boolean" + }, + "timestamp_detection_no_match_reason": { + "type": [ + "string", + "null" + ] + } + }, + "required": [ + "timestamp_detection_method", + "timestamp_detection_model_called", + "timestamp_detection_no_match_reason", + "source_inference" + ], + "type": "object" + }, + "duplicate": { + "additionalProperties": true, + "properties": { + "entry_count": { + "type": [ + "integer", + "null" + ] + }, + "import_id": { + "type": "string" + }, + "imported_at": { + "type": [ + "string", + "null" + ] + }, + "state": { + "enum": [ + "imported", + "staged" + ], + "type": "string" + } + }, + "required": [ + "import_id", + "imported_at", + "entry_count", + "state" + ], + "type": "object" + }, "facet": { "type": [ "string", "null" ] }, + "metadata": { + "additionalProperties": true, + "properties": { + "client": { + "additionalProperties": true, + "type": "object" + }, + "imported_via": { + "type": [ + "string", + "null" + ] + }, + "mime_type": { + "type": [ + "string", + "null" + ] + }, + "observer_handle": { + "type": [ + "string", + "null" + ] + }, + "original_filename": { + "type": [ + "string", + "null" + ] + }, + "source_hint": { + "type": [ + "string", + "null" + ] + } + }, + "required": [ + "original_filename", + "mime_type", + "imported_via", + "observer_handle", + "source_hint", + "client" + ], + "type": "object" + }, "path": { "type": "string" }, + "recommended_action": { + "enum": [ + "start", + "do_not_start" + ], + "type": "string" + }, + "replay": { + "type": "boolean" + }, + "schema_version": { + "type": "integer" + }, "setting": { "type": [ "string", "null" ] }, + "source": { + "enum": [ + "audio", + "image", + "document", + "text" + ], + "type": "string" + }, + "status": { + "enum": [ + "staged", + "duplicate" + ], + "type": "string" + }, "timestamp": { "type": "string" + } + }, + "required": [ + "schema_version", + "status", + "replay", + "path", + "timestamp", + "client_item_id", + "source", + "facet", + "setting", + "recommended_action", + "metadata", + "diagnostics" + ], + "type": "object" + } + } + }, + "description": "Import source staged, replayed, or identified as a duplicate." + }, + "400": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/Error" + } + } + }, + "description": "Required fields were missing or neither file nor text was supplied.", + "x-reason-codes": [ + "ingest_no_files", + "missing_required_field" + ] + }, + "403": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/Error" + } + } + }, + "description": "Access gate rejected a revoked paired-link identity.", + "x-reason-codes": [ + "pl_revoked" + ] + }, + "409": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/Error" + } + } + }, + "description": "client_item_id already names different staged content.", + "x-reason-codes": [ + "import_client_id_conflict" + ] + } + }, + "summary": "Save import source", + "tags": [ + "import" + ] + } + }, + "/app/import/api/save-path": { + "post": { + "description": "Register a local filesystem path for import staging using the same idempotent summary response as import.save.", + "operationId": "import.savePath", + "requestBody": { + "content": { + "application/json": { + "example": { + "client": {}, + "client_item_id": "ios-path-1357", + "path": "/Users/sol/Documents/Notes", + "source_hint": "obsidian" + }, + "schema": { + "additionalProperties": true, + "properties": { + "client": { + "type": "object" + }, + "client_item_id": { + "type": "string" + }, + "facet": { + "type": "string" + }, + "imported_via": { + "type": "string" + }, + "observer_handle": { + "type": "string" + }, + "path": { + "type": "string" + }, + "setting": { + "type": "string" + }, + "source_hint": { + "type": "string" + } + }, + "required": [ + "client_item_id", + "path" + ], + "type": "object" + } + } + }, + "required": true + }, + "responses": { + "200": { + "content": { + "application/json": { + "example": { + "client_item_id": "ios-path-1357", + "diagnostics": { + "source_inference": "extension", + "timestamp_detection_method": "upload_fallback", + "timestamp_detection_model_called": false, + "timestamp_detection_no_match_reason": null + }, + "facet": null, + "metadata": { + "client": {}, + "imported_via": "ios", + "mime_type": null, + "observer_handle": null, + "original_filename": "Notes", + "source_hint": "obsidian" + }, + "path": "/Users/sol/Documents/Notes", + "recommended_action": "start", + "replay": false, + "schema_version": 1, + "setting": null, + "source": "text", + "status": "staged", + "timestamp": "20260618_143022" + }, + "schema": { + "additionalProperties": true, + "properties": { + "client_item_id": { + "type": "string" }, - "timestamp_detection_method": { - "description": "Timestamp detection method: deterministic, model, upload_fallback, or explicit.", + "diagnostics": { + "additionalProperties": true, + "properties": { + "source_inference": { + "enum": [ + "extension", + "content_type", + "default" + ], + "type": "string" + }, + "timestamp_detection_method": { + "type": "string" + }, + "timestamp_detection_model_called": { + "type": "boolean" + }, + "timestamp_detection_no_match_reason": { + "type": [ + "string", + "null" + ] + } + }, + "required": [ + "timestamp_detection_method", + "timestamp_detection_model_called", + "timestamp_detection_no_match_reason", + "source_inference" + ], + "type": "object" + }, + "duplicate": { + "additionalProperties": true, + "properties": { + "entry_count": { + "type": [ + "integer", + "null" + ] + }, + "import_id": { + "type": "string" + }, + "imported_at": { + "type": [ + "string", + "null" + ] + }, + "state": { + "enum": [ + "imported", + "staged" + ], + "type": "string" + } + }, + "required": [ + "import_id", + "imported_at", + "entry_count", + "state" + ], + "type": "object" + }, + "facet": { + "type": [ + "string", + "null" + ] + }, + "metadata": { + "additionalProperties": true, + "properties": { + "client": { + "additionalProperties": true, + "type": "object" + }, + "imported_via": { + "type": [ + "string", + "null" + ] + }, + "mime_type": { + "type": [ + "string", + "null" + ] + }, + "observer_handle": { + "type": [ + "string", + "null" + ] + }, + "original_filename": { + "type": [ + "string", + "null" + ] + }, + "source_hint": { + "type": [ + "string", + "null" + ] + } + }, + "required": [ + "original_filename", + "mime_type", + "imported_via", + "observer_handle", + "source_hint", + "client" + ], + "type": "object" + }, + "path": { "type": "string" }, - "timestamp_detection_model_called": { + "recommended_action": { + "enum": [ + "start", + "do_not_start" + ], + "type": "string" + }, + "replay": { "type": "boolean" }, - "timestamp_detection_no_match_reason": { + "schema_version": { + "type": "integer" + }, + "setting": { "type": [ "string", "null" ] + }, + "source": { + "enum": [ + "audio", + "image", + "document", + "text" + ], + "type": "string" + }, + "status": { + "enum": [ + "staged", + "duplicate" + ], + "type": "string" + }, + "timestamp": { + "type": "string" } }, "required": [ + "schema_version", + "status", + "replay", "path", "timestamp", + "client_item_id", + "source", "facet", "setting", - "timestamp_detection_method", - "timestamp_detection_model_called", - "timestamp_detection_no_match_reason" + "recommended_action", + "metadata", + "diagnostics" ], "type": "object" } } }, - "description": "Import source saved." + "description": "Import path staged, replayed, or identified as a duplicate." }, "400": { "content": { @@ -1414,9 +2050,9 @@ } } }, - "description": "Neither file nor text was supplied.", + "description": "client_item_id or path was missing.", "x-reason-codes": [ - "ingest_no_files" + "missing_required_field" ] }, "403": { @@ -1431,9 +2067,35 @@ "x-reason-codes": [ "pl_revoked" ] + }, + "404": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/Error" + } + } + }, + "description": "The local path did not exist.", + "x-reason-codes": [ + "file_not_found" + ] + }, + "409": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/Error" + } + } + }, + "description": "client_item_id already names different staged content.", + "x-reason-codes": [ + "import_client_id_conflict" + ] } }, - "summary": "Save import source", + "summary": "Save import source path", "tags": [ "import" ] @@ -1441,7 +2103,7 @@ }, "/app/import/api/start": { "post": { - "description": "Start processing a previously saved import source.", + "description": "Start processing a previously saved import source. Saved import metadata is authoritative for facet, setting, and source routing.", "operationId": "import.start", "requestBody": { "content": { @@ -1449,7 +2111,6 @@ "example": { "force": false, "path": "/journal/imports/20260618_143022/source.txt", - "source": "manual", "timestamp": "20260618_143022" }, "schema": { @@ -1461,9 +2122,6 @@ "path": { "type": "string" }, - "source": { - "type": "string" - }, "timestamp": { "type": "string" } @@ -1514,8 +2172,9 @@ } } }, - "description": "Path or timestamp was missing.", + "description": "Path or timestamp was missing, or the import is terminal.", "x-reason-codes": [ + "invalid_operation_for_state", "missing_required_field" ] }, diff --git a/solstone/apps/import/contract.py b/solstone/apps/import/contract.py index 94491c625..70c9e752b 100644 --- a/solstone/apps/import/contract.py +++ b/solstone/apps/import/contract.py @@ -21,6 +21,110 @@ def _json_error( _NULLABLE_STRING = {"type": ["string", "null"]} +_NULLABLE_INTEGER = {"type": ["integer", "null"]} +_FREE_OBJECT = {"type": "object", "additionalProperties": True} +_STATUS_SCHEMA = {"type": "string", "enum": ["staged", "duplicate"]} +_SOURCE_SCHEMA = {"type": "string", "enum": ["audio", "image", "document", "text"]} +_ACTION_SCHEMA = {"type": "string", "enum": ["start", "do_not_start"]} +_SOURCE_INFERENCE_SCHEMA = { + "type": "string", + "enum": ["extension", "content_type", "default"], +} +_METADATA_SCHEMA = { + "type": "object", + "additionalProperties": True, + "properties": { + "original_filename": _NULLABLE_STRING, + "mime_type": _NULLABLE_STRING, + "imported_via": _NULLABLE_STRING, + "observer_handle": _NULLABLE_STRING, + "source_hint": _NULLABLE_STRING, + "client": _FREE_OBJECT, + }, + "required": [ + "original_filename", + "mime_type", + "imported_via", + "observer_handle", + "source_hint", + "client", + ], +} +_DIAGNOSTICS_SCHEMA = { + "type": "object", + "additionalProperties": True, + "properties": { + "timestamp_detection_method": {"type": "string"}, + "timestamp_detection_model_called": {"type": "boolean"}, + "timestamp_detection_no_match_reason": _NULLABLE_STRING, + "source_inference": _SOURCE_INFERENCE_SCHEMA, + }, + "required": [ + "timestamp_detection_method", + "timestamp_detection_model_called", + "timestamp_detection_no_match_reason", + "source_inference", + ], +} +_DUPLICATE_SCHEMA = { + "type": "object", + "additionalProperties": True, + "properties": { + "import_id": {"type": "string"}, + "imported_at": _NULLABLE_STRING, + "entry_count": _NULLABLE_INTEGER, + "state": {"type": "string", "enum": ["imported", "staged"]}, + }, + "required": ["import_id", "imported_at", "entry_count", "state"], +} + +_SAVE_RESPONSE_FIELDS = ( + FieldSpec("schema_version", "integer", required=True), + FieldSpec("status", "string", required=True, raw_schema=_STATUS_SCHEMA), + FieldSpec("replay", "boolean", required=True), + FieldSpec("path", "string", required=True), + FieldSpec("timestamp", "string", required=True), + FieldSpec("client_item_id", "string", required=True), + FieldSpec("source", "string", required=True, raw_schema=_SOURCE_SCHEMA), + FieldSpec("facet", "string", required=True, raw_schema=_NULLABLE_STRING), + FieldSpec("setting", "string", required=True, raw_schema=_NULLABLE_STRING), + FieldSpec( + "recommended_action", + "string", + required=True, + raw_schema=_ACTION_SCHEMA, + ), + FieldSpec("metadata", "object", required=True, raw_schema=_METADATA_SCHEMA), + FieldSpec("diagnostics", "object", required=True, raw_schema=_DIAGNOSTICS_SCHEMA), + FieldSpec("duplicate", "object", raw_schema=_DUPLICATE_SCHEMA), +) + +_SAVE_RESPONSE_EXAMPLE = { + "schema_version": 1, + "status": "staged", + "replay": False, + "path": "/journal/imports/20260618_143022/source.m4a", + "timestamp": "20260618_143022", + "client_item_id": "ios-item-4f8b", + "source": "audio", + "facet": None, + "setting": None, + "recommended_action": "start", + "metadata": { + "original_filename": "source.m4a", + "mime_type": "audio/mp4", + "imported_via": "ios", + "observer_handle": None, + "source_hint": None, + "client": {}, + }, + "diagnostics": { + "timestamp_detection_method": "upload_fallback", + "timestamp_detection_model_called": False, + "timestamp_detection_no_match_reason": None, + "source_inference": "extension", + }, +} OPERATIONS: list[OperationSpec] = [ @@ -31,11 +135,13 @@ OPERATIONS: list[OperationSpec] = [ summary="Save import source", description=( "Save an uploaded import file or pasted text into imports staging. " - "Submit either file or text." + "Submit either file or text. client_item_id is required for " + "idempotent native-client staging." ), request=RequestSpec( content_type="multipart/form-data", fields=( + FieldSpec("client_item_id", "string", required=True), FieldSpec( "file", "string", @@ -44,66 +150,163 @@ OPERATIONS: list[OperationSpec] = [ FieldSpec("text", "string"), FieldSpec("facet", "string"), FieldSpec("setting", "string"), + FieldSpec("source_hint", "string"), FieldSpec("imported_via", "string"), FieldSpec("observer_handle", "string"), FieldSpec("deterministic_only", "boolean"), + FieldSpec("client", "object"), ), description="Multipart body with either file or text.", ), responses=( ResponseSpec( status=200, - description="Import source saved.", + description=( + "Import source staged, replayed, or identified as a duplicate." + ), + named_fields=_SAVE_RESPONSE_FIELDS, + example=_SAVE_RESPONSE_EXAMPLE, + ), + _json_error( + 400, + ("ingest_no_files", "missing_required_field"), + "Required fields were missing or neither file nor text was supplied.", + ), + _json_error( + 409, + ("import_client_id_conflict",), + "client_item_id already names different staged content.", + ), + _json_error( + 403, + ("pl_revoked",), + "Access gate rejected a revoked paired-link identity.", + ), + ), + ), + OperationSpec( + operation_id="import.savePath", + method="POST", + rule="/app/import/api/save-path", + summary="Save import source path", + description=( + "Register a local filesystem path for import staging using the same " + "idempotent summary response as import.save." + ), + request=RequestSpec( + fields=( + FieldSpec("client_item_id", "string", required=True), + FieldSpec("path", "string", required=True), + FieldSpec("facet", "string"), + FieldSpec("setting", "string"), + FieldSpec("source_hint", "string"), + FieldSpec("imported_via", "string"), + FieldSpec("observer_handle", "string"), + FieldSpec("client", "object"), + ), + example={ + "client_item_id": "ios-path-1357", + "path": "/Users/sol/Documents/Notes", + "source_hint": "obsidian", + "client": {}, + }, + ), + responses=( + ResponseSpec( + status=200, + description=( + "Import path staged, replayed, or identified as a duplicate." + ), + named_fields=_SAVE_RESPONSE_FIELDS, + example={ + **_SAVE_RESPONSE_EXAMPLE, + "path": "/Users/sol/Documents/Notes", + "client_item_id": "ios-path-1357", + "source": "text", + "metadata": { + **_SAVE_RESPONSE_EXAMPLE["metadata"], + "original_filename": "Notes", + "mime_type": None, + "source_hint": "obsidian", + }, + }, + ), + _json_error( + 400, + ("missing_required_field",), + "client_item_id or path was missing.", + ), + _json_error( + 404, + ("file_not_found",), + "The local path did not exist.", + ), + _json_error( + 409, + ("import_client_id_conflict",), + "client_item_id already names different staged content.", + ), + _json_error( + 403, + ("pl_revoked",), + "Access gate rejected a revoked paired-link identity.", + ), + ), + ), + OperationSpec( + operation_id="import.meta", + method="POST", + rule="/app/import/api/meta", + summary="Update import metadata", + description="Update allowlisted metadata fields on a not-yet-started import.", + request=RequestSpec( + fields=( + FieldSpec("path", "string", required=True), + FieldSpec("facet", "string"), + FieldSpec("setting", "string"), + FieldSpec("original_filename", "string"), + FieldSpec("mime_type", "string"), + FieldSpec("source_hint", "string"), + FieldSpec("observer_handle", "string"), + FieldSpec("imported_via", "string"), + FieldSpec("client", "object"), + ), + example={ + "path": "/journal/imports/20260618_143022/source.m4a", + "facet": "work", + }, + ), + responses=( + ResponseSpec( + status=200, + description="Import metadata updated.", named_fields=( + FieldSpec("status", "string", required=True), FieldSpec("path", "string", required=True), FieldSpec("timestamp", "string", required=True), - FieldSpec( - "facet", - "string", - required=True, - raw_schema=_NULLABLE_STRING, - ), - FieldSpec( - "setting", - "string", - required=True, - raw_schema=_NULLABLE_STRING, - ), - FieldSpec( - "timestamp_detection_method", - "string", - required=True, - description=( - "Timestamp detection method: deterministic, model, " - "upload_fallback, or explicit." - ), - ), - FieldSpec( - "timestamp_detection_model_called", - "boolean", - required=True, - ), - FieldSpec( - "timestamp_detection_no_match_reason", - "string", - required=True, - raw_schema=_NULLABLE_STRING, - ), + FieldSpec("updated", "object", required=True), ), example={ - "path": "/journal/imports/20260618_143022/source.txt", + "status": "ok", + "path": "/journal/imports/20260618_143022/source.m4a", "timestamp": "20260618_143022", - "facet": None, - "setting": None, - "timestamp_detection_method": "deterministic", - "timestamp_detection_model_called": False, - "timestamp_detection_no_match_reason": None, + "updated": {"facet": "work"}, }, ), _json_error( 400, - ("ingest_no_files",), - "Neither file nor text was supplied.", + ("invalid_operation_for_state", "missing_required_field"), + "The import path was missing or the import state is terminal.", + ), + _json_error( + 404, + ("import_not_found",), + "Import metadata was not found.", + ), + _json_error( + 500, + ("import_metadata_failed",), + "Import metadata could not be read or updated.", ), _json_error( 403, @@ -117,18 +320,19 @@ OPERATIONS: list[OperationSpec] = [ method="POST", rule="/app/import/api/start", summary="Start import", - description="Start processing a previously saved import source.", + description=( + "Start processing a previously saved import source. Saved import " + "metadata is authoritative for facet, setting, and source routing." + ), request=RequestSpec( fields=( FieldSpec("path", "string", required=True), FieldSpec("timestamp", "string", required=True), - FieldSpec("source", "string"), FieldSpec("force", "boolean"), ), example={ "path": "/journal/imports/20260618_143022/source.txt", "timestamp": "20260618_143022", - "source": "manual", "force": False, }, ), @@ -144,8 +348,8 @@ OPERATIONS: list[OperationSpec] = [ ), _json_error( 400, - ("missing_required_field",), - "Path or timestamp was missing.", + ("invalid_operation_for_state", "missing_required_field"), + "Path or timestamp was missing, or the import is terminal.", ), _json_error( 404, diff --git a/solstone/apps/import/routes.py b/solstone/apps/import/routes.py index b9b811119..da52f55ca 100644 --- a/solstone/apps/import/routes.py +++ b/solstone/apps/import/routes.py @@ -4,9 +4,9 @@ from __future__ import annotations import json -import logging import re import time +from datetime import datetime from pathlib import Path from typing import Any @@ -17,10 +17,12 @@ from solstone.apps.utils import log_app_action from solstone.convey import emit, state from solstone.convey.reasons import ( FILE_NOT_FOUND, + IMPORT_CLIENT_ID_CONFLICT, IMPORT_CONFLICT, IMPORT_METADATA_FAILED, IMPORT_NOT_FOUND, INGEST_NO_FILES, + INVALID_OPERATION_FOR_STATE, INVALID_REQUEST_VALUE, JOURNAL_SOURCE_PROBLEM, MISSING_REQUIRED_FIELD, @@ -32,8 +34,11 @@ from solstone.convey.utils import ( success_response, ) from solstone.think.detect_created import detect_created, resolve_created_deterministic +from solstone.think.importers.shared import find_manifest_by_hash, hash_source from solstone.think.importers.utils import ( build_import_info, + find_staged_by_client_item_id, + find_staged_by_source_hash, generate_content_manifest, get_import_details, list_import_timestamps, @@ -44,7 +49,11 @@ from solstone.think.importers.utils import ( update_import_metadata_fields, write_import_metadata, ) -from solstone.think.media import MEDIA_EXTENSIONS +from solstone.think.media import ( + MEDIA_EXTENSIONS, + canonical_source, + canonical_source_signal, +) from solstone.think.utils import day_path, now_ms from .journal_sources import ( @@ -222,15 +231,233 @@ def _form_bool(value: str | None) -> bool: return value.strip().lower() in {"true", "1", "yes"} if value else False +CANONICAL_IMPORT_SOURCES = {"audio", "image", "document", "text"} + + +def _clean_optional(value: Any) -> str | None: + if value is None: + return None + if isinstance(value, str): + cleaned = value.strip() + else: + cleaned = str(value).strip() + return cleaned or None + + +def _client_bag(value: Any) -> dict: + if isinstance(value, dict): + return value + if isinstance(value, str) and value.strip(): + try: + parsed = json.loads(value) + except json.JSONDecodeError: + return {} + return parsed if isinstance(parsed, dict) else {} + return {} + + +def _build_save_summary( + metadata: dict, + *, + status: str, + replay: bool, + duplicate: dict | None, + recommended_action: str | None = None, +) -> dict: + """Build the versioned import staging summary response.""" + client = metadata.get("client") + if not isinstance(client, dict): + client = {} + action = recommended_action + if action is None: + action = "do_not_start" if status == "duplicate" else "start" + + summary: dict[str, Any] = { + "schema_version": 1, + "status": status, + "replay": replay, + "path": str(metadata.get("file_path", "")), + "timestamp": str(metadata.get("user_timestamp", "")), + "client_item_id": str(metadata.get("client_item_id", "")), + "source": metadata.get("source", "text"), + "facet": metadata.get("facet"), + "setting": metadata.get("setting"), + "recommended_action": action, + "metadata": { + "original_filename": metadata.get("original_filename"), + "mime_type": metadata.get("mime_type"), + "imported_via": metadata.get("imported_via"), + "observer_handle": metadata.get("observer_handle"), + "source_hint": metadata.get("source_hint"), + "client": client, + }, + "diagnostics": { + "timestamp_detection_method": metadata.get( + "timestamp_detection_method", "duplicate" + ), + "timestamp_detection_model_called": metadata.get( + "timestamp_detection_model_called", False + ), + "timestamp_detection_no_match_reason": metadata.get( + "timestamp_detection_no_match_reason" + ), + "source_inference": metadata.get("source_inference", "default"), + }, + } + if duplicate is not None: + summary["duplicate"] = { + "import_id": duplicate.get("import_id"), + "imported_at": duplicate.get("imported_at"), + "entry_count": duplicate.get("entry_count"), + "state": duplicate.get("state"), + } + return summary + + +def _load_import_metadata_or_none(journal_root: Path, timestamp: str) -> dict | None: + try: + metadata = read_import_metadata(journal_root, timestamp) + except (FileNotFoundError, json.JSONDecodeError, OSError): + return None + return metadata if isinstance(metadata, dict) else None + + +def _duplicate_summary_metadata( + journal_root: Path, + *, + client_item_id: str, + source: str, + source_inference: str, + duplicate: dict, + existing_metadata: dict | None = None, +) -> dict: + import_id = str(duplicate.get("import_id") or "") + metadata = dict(existing_metadata or {}) + if not metadata and import_id: + metadata = dict(_load_import_metadata_or_none(journal_root, import_id) or {}) + + metadata.setdefault("file_path", str(journal_root / "imports" / import_id)) + metadata.setdefault("user_timestamp", import_id) + metadata.setdefault("original_filename", None) + metadata.setdefault("mime_type", None) + metadata.setdefault("imported_via", duplicate.get("imported_via")) + metadata.setdefault("observer_handle", duplicate.get("observer_handle")) + metadata.setdefault("source_hint", None) + metadata.setdefault("client", {}) + metadata.setdefault("facet", None) + metadata.setdefault("setting", None) + metadata.setdefault("timestamp_detection_method", "duplicate") + metadata.setdefault("timestamp_detection_model_called", False) + metadata.setdefault("timestamp_detection_no_match_reason", None) + metadata["client_item_id"] = client_item_id + metadata["source_inference"] = metadata.get("source_inference") or source_inference + if metadata.get("source") not in CANONICAL_IMPORT_SOURCES: + metadata["source"] = source + return metadata + + +def _duplicate_or_replay_response( + journal_root: Path, + *, + client_item_id: str, + source_hash: str, + source: str, + source_inference: str, +) -> Any | None: + existing = find_staged_by_client_item_id(journal_root, client_item_id) + if existing: + if existing.get("source_hash") == source_hash: + replay_action = ( + "do_not_start" + if existing.get("task_id") or existing.get("processing_completed") + else None + ) + return jsonify( + _build_save_summary( + existing, + status="staged", + replay=True, + duplicate=None, + recommended_action=replay_action, + ) + ) + return error_response( + IMPORT_CLIENT_ID_CONFLICT, + detail=( + "client_item_id already staged for different content; use a new " + "client_item_id or re-fetch the existing item" + ), + ) + + imported = find_manifest_by_hash(journal_root, source_hash) + if imported: + duplicate = { + "import_id": imported.get("import_id"), + "imported_at": imported.get("imported_at"), + "entry_count": imported.get("entry_count"), + "state": "imported", + "imported_via": imported.get("imported_via"), + "observer_handle": imported.get("observer_handle"), + } + metadata = _duplicate_summary_metadata( + journal_root, + client_item_id=client_item_id, + source=source, + source_inference=source_inference, + duplicate=duplicate, + ) + return jsonify( + _build_save_summary( + metadata, + status="duplicate", + replay=False, + duplicate=duplicate, + ) + ) + + staged_duplicate = find_staged_by_source_hash(journal_root, source_hash) + if staged_duplicate: + duplicate = { + "import_id": staged_duplicate.get("timestamp"), + "imported_at": None, + "entry_count": None, + "state": "staged", + } + metadata = _duplicate_summary_metadata( + journal_root, + client_item_id=client_item_id, + source=source, + source_inference=source_inference, + duplicate=duplicate, + existing_metadata=staged_duplicate, + ) + return jsonify( + _build_save_summary( + metadata, + status="duplicate", + replay=False, + duplicate=duplicate, + ) + ) + + return None + + @import_bp.route("/api/save", methods=["POST"]) def import_save() -> Any: - from datetime import datetime - upload = request.files.get("file") text = request.form.get("text", "").strip() + client_item_id = request.form.get("client_item_id", "").strip() facet = request.form.get("facet", "").strip() or None setting = request.form.get("setting", "").strip() or None + source_hint = request.form.get("source_hint", "").strip() or None + imported_via = request.form.get("imported_via", "").strip() or "web_dashboard" + observer_handle = request.form.get("observer_handle", "").strip() or None deterministic_only = _form_bool(request.form.get("deterministic_only")) + client = _client_bag(request.form.get("client")) + + if not client_item_id: + return error_response(MISSING_REQUIRED_FIELD, detail="Missing client_item_id") # Generate timestamp for folder name timestamp_ms = now_ms() @@ -243,6 +470,15 @@ def import_save() -> Any: else: return error_response(INGEST_NO_FILES, detail="No input") + original_filename = upload.filename if upload else "paste.txt" + mime_type = upload.content_type if upload else "text/plain" + source = canonical_source(filename=original_filename, content_type=mime_type) + source_inference = canonical_source_signal( + filename=original_filename, + content_type=mime_type, + ) + journal_root = Path(state.journal_root) + # Detect timestamp from content first (need temporary save for detection) ts = None detection_result = None @@ -272,6 +508,18 @@ def import_save() -> Any: temp_path = tmp.name try: + temp_source = Path(temp_path) + source_hash = hash_source(temp_source) + duplicate_or_replay = _duplicate_or_replay_response( + journal_root, + client_item_id=client_item_id, + source_hash=source_hash, + source=source, + source_inference=source_inference, + ) + if duplicate_or_replay is not None: + return duplicate_or_replay + try: original_name = upload.filename if upload else None detection_result = resolve_created_deterministic( @@ -313,111 +561,92 @@ def import_save() -> Any: detection_result = None timestamp_detection_model_called = True timestamp_detection_no_match_reason = "model_no_match" - finally: - # Clean up temporary file - Path(temp_path).unlink(missing_ok=True) - - # Use detected timestamp or fall back to upload timestamp - folder_timestamp = ( - ts - if ts - else f"{datetime.fromtimestamp(timestamp_ms / 1000).strftime('%Y%m%d_%H%M%S')}" - ) - - # Save the actual file using utility function - if upload: - # Save uploaded file to temp location first, then move to import dir - import tempfile - with tempfile.NamedTemporaryFile(delete=False) as tmp: - upload.save(tmp.name) - temp_source = Path(tmp.name) + # Use detected timestamp or fall back to upload timestamp + folder_timestamp = ( + ts + if ts + else datetime.fromtimestamp(timestamp_ms / 1000).strftime("%Y%m%d_%H%M%S") + ) - try: + # Save the actual file using utility function + if upload: file_path = save_import_file( - journal_root=Path(state.journal_root), + journal_root=journal_root, timestamp=folder_timestamp, source_path=temp_source, filename=filename, ) - finally: - temp_source.unlink(missing_ok=True) - else: - file_path = save_import_text( - journal_root=Path(state.journal_root), - timestamp=folder_timestamp, - content=text, - filename=filename, - ) - - # Build metadata dict - metadata = { - "original_filename": upload.filename if upload else "paste.txt", - "upload_timestamp": timestamp_ms, - "upload_datetime": datetime.fromtimestamp(timestamp_ms / 1000).isoformat(), - "detection_result": detection_result, - "detected_timestamp": ts, - "user_timestamp": folder_timestamp, # The timestamp used for the folder - "timestamp_detection_method": timestamp_detection_method, - "timestamp_detection_model_called": timestamp_detection_model_called, - "timestamp_detection_no_match_reason": timestamp_detection_no_match_reason, - "file_size": file_path.stat().st_size if file_path.exists() else 0, - "mime_type": upload.content_type if upload else "text/plain", - "facet": facet, # Include selected facet - "setting": setting, - "file_path": str(file_path), # Store the actual file path - "imported_via": request.form.get("imported_via", "").strip() or "web_dashboard", - "link_id": _link_id_from_identity(), - "observer_handle": request.form.get("observer_handle", "").strip() or None, - } - - # Write metadata using utility function - write_import_metadata( - journal_root=Path(state.journal_root), - timestamp=folder_timestamp, - metadata=metadata, - ) + else: + file_path = save_import_text( + journal_root=journal_root, + timestamp=folder_timestamp, + content=text, + filename=filename, + ) - # Check for dedup — has this exact file been imported before? - dedup = None - try: - from solstone.think.importers.shared import find_manifest_by_hash, hash_source - - source_hash = hash_source(file_path) - existing = find_manifest_by_hash(Path(state.journal_root), source_hash) - if existing: - dedup = { - "imported_at": existing.get("imported_at", "unknown"), - "entry_count": existing.get("entry_count", 0), - "import_id": existing.get("import_id", ""), - } - except OSError as exc: - logging.warning("Dedup check failed for %s: %s", file_path, exc) + # Build metadata dict + metadata = { + "original_filename": original_filename, + "upload_timestamp": timestamp_ms, + "upload_datetime": datetime.fromtimestamp(timestamp_ms / 1000).isoformat(), + "detection_result": detection_result, + "detected_timestamp": ts, + "user_timestamp": folder_timestamp, + "timestamp_detection_method": timestamp_detection_method, + "timestamp_detection_model_called": timestamp_detection_model_called, + "timestamp_detection_no_match_reason": timestamp_detection_no_match_reason, + "source_inference": source_inference, + "file_size": file_path.stat().st_size if file_path.exists() else 0, + "mime_type": mime_type, + "facet": facet, + "setting": setting, + "file_path": str(file_path), + "imported_via": imported_via, + "link_id": _link_id_from_identity(), + "observer_handle": observer_handle, + "client_item_id": client_item_id, + "source_hash": source_hash, + "source": source, + "source_hint": source_hint, + "client": client, + } - result: dict[str, Any] = { - "path": str(file_path), - "timestamp": folder_timestamp, - "facet": facet, - "setting": setting, - "timestamp_detection_method": timestamp_detection_method, - "timestamp_detection_model_called": timestamp_detection_model_called, - "timestamp_detection_no_match_reason": timestamp_detection_no_match_reason, - } - if dedup: - result["dedup"] = dedup + # Write metadata using utility function + write_import_metadata( + journal_root=journal_root, + timestamp=folder_timestamp, + metadata=metadata, + ) - return jsonify(result) + return jsonify( + _build_save_summary( + metadata, + status="staged", + replay=False, + duplicate=None, + ) + ) + finally: + # Clean up temporary file + Path(temp_path).unlink(missing_ok=True) @import_bp.route("/api/save-path", methods=["POST"]) def import_save_path() -> Any: """Register a local filesystem path for import (e.g. Obsidian vault).""" - from datetime import datetime - data = request.get_json(force=True) + client_item_id = data.get("client_item_id", "").strip() local_path = data.get("path", "").strip() facet = data.get("facet", "").strip() or None setting = data.get("setting", "").strip() or None + source_hint = data.get("source_hint", "").strip() or None + imported_via = data.get("imported_via", "").strip() or "web_dashboard" + observer_handle = data.get("observer_handle", "").strip() or None + client = _client_bag(data.get("client")) + + if not client_item_id: + return error_response(MISSING_REQUIRED_FIELD, detail="Missing client_item_id") if not local_path: return error_response(MISSING_REQUIRED_FIELD, detail="Missing path") @@ -431,23 +660,42 @@ def import_save_path() -> Any: f"{datetime.fromtimestamp(timestamp_ms / 1000).strftime('%Y%m%d_%H%M%S')}" ) - # Create import directory and metadata journal_root = Path(state.journal_root) - import_dir = journal_root / "imports" / folder_timestamp - import_dir.mkdir(parents=True, exist_ok=True) + source_hash = hash_source(local) + source = canonical_source(filename=local.name) + source_inference = canonical_source_signal(filename=local.name) + duplicate_or_replay = _duplicate_or_replay_response( + journal_root, + client_item_id=client_item_id, + source_hash=source_hash, + source=source, + source_inference=source_inference, + ) + if duplicate_or_replay is not None: + return duplicate_or_replay metadata = { "original_filename": local.name, "upload_timestamp": timestamp_ms, "upload_datetime": datetime.fromtimestamp(timestamp_ms / 1000).isoformat(), "user_timestamp": folder_timestamp, + "timestamp_detection_method": "path_fallback", + "timestamp_detection_model_called": False, + "timestamp_detection_no_match_reason": None, + "source_inference": source_inference, "file_path": local_path, "facet": facet, "setting": setting, "is_local_path": True, - "imported_via": data.get("imported_via", "").strip() or "web_dashboard", + "mime_type": None, + "imported_via": imported_via, "link_id": _link_id_from_identity(), - "observer_handle": data.get("observer_handle", "").strip() or None, + "observer_handle": observer_handle, + "client_item_id": client_item_id, + "source_hash": source_hash, + "source": source, + "source_hint": source_hint, + "client": client, } write_import_metadata( @@ -457,37 +705,83 @@ def import_save_path() -> Any: ) return jsonify( - { - "path": local_path, - "timestamp": folder_timestamp, - "facet": facet, - "setting": setting, - } + _build_save_summary( + metadata, + status="staged", + replay=False, + duplicate=None, + ) ) -@import_bp.route("/api/facet", methods=["POST"]) +@import_bp.route("/api/meta", methods=["POST"]) def import_update_metadata() -> Any: - """Update stored metadata (facet/setting) for a saved import.""" + """Update stored metadata for a saved import.""" data = request.get_json(force=True) raw_path = data.get("path", "").strip() if not raw_path: return error_response(MISSING_REQUIRED_FIELD, detail="Missing import path") - facet = data.get("facet", "").strip() or None - setting = data.get("setting", "").strip() or None - # Extract timestamp from path # Path format: .../imports/{timestamp}/{filename} file_path = Path(raw_path) timestamp = file_path.parent.name + journal_root = Path(state.journal_root) try: - # Use utility function to update metadata - metadata, updated = update_import_metadata_fields( - journal_root=Path(state.journal_root), + metadata = read_import_metadata(journal_root=journal_root, timestamp=timestamp) + except FileNotFoundError: + return error_response(IMPORT_NOT_FOUND, detail="Import metadata not found") + except Exception as exc: + return error_response( + IMPORT_METADATA_FAILED, + detail=f"Failed to read metadata: {exc}", + ) + + if metadata.get("task_id") or metadata.get("processing_completed"): + return error_response( + INVALID_OPERATION_FOR_STATE, + detail="import already started or processed", + ) + + source_hash = metadata.get("source_hash") + if source_hash and find_manifest_by_hash(journal_root, source_hash): + return error_response( + INVALID_OPERATION_FOR_STATE, + detail="content already imported", + ) + + updates: dict[str, Any] = {} + for key in ( + "facet", + "setting", + "original_filename", + "mime_type", + "source_hint", + "observer_handle", + "imported_via", + "client", + ): + if key not in data: + continue + if key in {"facet", "setting", "source_hint", "observer_handle"}: + updates[key] = _clean_optional(data.get(key)) + elif key == "client": + updates[key] = _client_bag(data.get(key)) + else: + updates[key] = data.get(key) + + changed = { + key: value + for key, value in updates.items() + if key not in metadata or metadata.get(key) != value + } + + try: + update_import_metadata_fields( + journal_root=journal_root, timestamp=timestamp, - updates={"facet": facet, "setting": setting}, + updates=updates, ) except FileNotFoundError: return error_response(IMPORT_NOT_FOUND, detail="Import metadata not found") @@ -500,9 +794,9 @@ def import_update_metadata() -> Any: return jsonify( { "status": "ok", - "facet": facet, - "setting": setting, - "updated": updated, + "path": raw_path, + "timestamp": timestamp, + "updated": changed, } ) @@ -854,14 +1148,10 @@ def import_start() -> Any: data = request.get_json(force=True) path = data.get("path") ts = data.get("timestamp") - source = data.get("source") force = data.get("force", False) if not path or not ts: return error_response(MISSING_REQUIRED_FIELD, detail="missing params") - # Generate task ID - task_id = str(now_ms()) - # Extract original timestamp from path and handle timestamp changes file_path = Path(path) journal_root = Path(state.journal_root) @@ -869,6 +1159,34 @@ def import_start() -> Any: is_local_path = not str(file_path).startswith(str(imports_dir)) original_timestamp = file_path.parent.name if not is_local_path else ts + # Read import metadata before any move. Saved metadata is the authority for + # facet, setting, and source routing. + try: + metadata = read_import_metadata( + journal_root=journal_root, + timestamp=original_timestamp, + ) + except FileNotFoundError: + return error_response( + IMPORT_NOT_FOUND, + detail=f"Import metadata not found for {original_timestamp}", + ) + except Exception as e: + return error_response( + IMPORT_METADATA_FAILED, + detail=f"Failed to read metadata: {str(e)}", + ) + + source_hash = metadata.get("source_hash") + if source_hash and find_manifest_by_hash(journal_root, source_hash): + return error_response( + INVALID_OPERATION_FOR_STATE, + detail="content already imported; will not start", + ) + + # Generate task ID + task_id = str(now_ms()) + # If timestamp changed, move the import directory through the imports/ owner if not is_local_path and original_timestamp != ts: try: @@ -899,20 +1217,6 @@ def import_start() -> Any: # Update file_path in metadata (need to update after reading) # We'll handle this after reading the metadata below - # Read import metadata to get facet and setting - try: - metadata = read_import_metadata(journal_root=journal_root, timestamp=ts) - except FileNotFoundError: - return error_response( - IMPORT_NOT_FOUND, - detail=f"Import metadata not found for {ts}", - ) - except Exception as e: - return error_response( - IMPORT_METADATA_FAILED, - detail=f"Failed to read metadata: {str(e)}", - ) - # Update file_path in metadata if timestamp changed if not is_local_path and original_timestamp != ts: try: @@ -921,6 +1225,7 @@ def import_start() -> Any: timestamp=ts, updates={"file_path": path}, ) + metadata["file_path"] = path except Exception as e: return error_response( IMPORT_METADATA_FAILED, @@ -929,6 +1234,7 @@ def import_start() -> Any: facet = metadata.get("facet") setting = metadata.get("setting") + source_hint = _clean_optional(metadata.get("source_hint")) # Build command cmd = ["journal", "importer", path, ts] @@ -936,8 +1242,8 @@ def import_start() -> Any: cmd.extend(["--facet", facet]) if setting: cmd.extend(["--setting", setting]) - if source: - cmd.extend(["--source", source]) + if source_hint: + cmd.extend(["--source", source_hint]) if force: cmd.append("--force") @@ -946,7 +1252,7 @@ def import_start() -> Any: update_import_metadata_fields( journal_root=journal_root, timestamp=ts, - updates={"task_id": task_id, "source": source}, + updates={"task_id": task_id, "source_hint": source_hint}, ) except Exception as e: return error_response( diff --git a/solstone/apps/import/workspace.html b/solstone/apps/import/workspace.html index 42f8f743a..54ba768cb 100644 --- a/solstone/apps/import/workspace.html +++ b/solstone/apps/import/workspace.html @@ -669,6 +669,7 @@ +
@@ -708,6 +709,9 @@ let currentFile = null; let currentGuideSource = null; let currentGuidedFile = null; let currentGuidedSaved = null; +window._quickClientItemId = null; +window._quickSaved = null; +window._guidedClientItemId = null; let importEvents = {}; let importsCache = []; let currentSourceFilter = ''; @@ -787,7 +791,7 @@ function showGrid() { currentGuideSource = null; currentGuidedFile = null; currentGuidedSaved = null; - window._guidedForceImport = false; + window._guidedClientItemId = null; navigateTo('grid'); } @@ -838,9 +842,42 @@ function showQuickImport() { let previousDetectFocus = null; let detectFocusTrapHandler = null; +function isTerminalDuplicate(data) { + return data?.status === 'duplicate' || data?.recommended_action === 'do_not_start'; +} + +function duplicateNoticeText(duplicate = {}) { + const state = duplicate.state || 'imported'; + const importId = duplicate.import_id || ''; + if (state === 'staged') { + return `this file was already staged${importId ? ` as ${importId}` : ''}.`; + } + const importedAt = duplicate.imported_at || 'an earlier import'; + const entries = duplicate.entry_count ? ` (${duplicate.entry_count} entries)` : ''; + return `this file was already imported on ${importedAt}${entries}.`; +} + function showDetect(res) { + window._quickSaved = res; document.getElementById('timestampInput').value = res.timestamp || ''; document.getElementById('savedPath').value = res.path; + const duplicateNotice = document.getElementById('detectDuplicateNotice'); + if (duplicateNotice) { + duplicateNotice.innerHTML = ''; + } + startBtn.style.display = ''; + startBtn.disabled = false; + if (isTerminalDuplicate(res)) { + if (duplicateNotice) { + duplicateNotice.innerHTML = ` +
+ ${escapeHtml(duplicateNoticeText(res.duplicate))} +
+ `; + } + startBtn.disabled = true; + startBtn.style.display = 'none'; + } const defaultFacet = res.facet || window.selectedFacet || ''; loadFacets(defaultFacet); settingInput.value = res.setting || ''; @@ -1406,6 +1443,7 @@ async function loadGuidedFlow(sourceName) { currentGuideSource = sourceName; currentGuidedFile = null; currentGuidedSaved = null; + window._guidedClientItemId = null; if (!guideSteps) { return; } @@ -1480,6 +1518,13 @@ async function loadGuidedFlow(sourceName) { } else { // Auto-detect default path for path_input sources checkDefaultPath(sourceName); + const guidedPathInput = document.getElementById('guidedPathInput'); + if (guidedPathInput) { + guidedPathInput.addEventListener('input', () => { + currentGuidedSaved = null; + window._guidedClientItemId = crypto.randomUUID(); + }); + } } document.getElementById('guidedStartBtn').addEventListener('click', () => startGuidedImport(source)); const guidedForm = document.getElementById('guidedImportForm'); @@ -1573,6 +1618,7 @@ function setupGuidedUploadArea() { currentGuidedFile = file || null; fileLabel.textContent = file ? file.name : ''; currentGuidedSaved = null; + window._guidedClientItemId = file ? crypto.randomUUID() : null; }; dropArea.onclick = () => fileInput.click(); @@ -1619,6 +1665,8 @@ function setupQuickImportForm() { } currentFile = null; + window._quickClientItemId = null; + window._quickSaved = null; dropArea.onclick = () => fileInput.click(); dropArea.addEventListener('keydown', e => { if (e.key === 'Enter' || e.key === ' ') { @@ -1639,14 +1687,24 @@ function setupQuickImportForm() { if (file) { currentFile = file; fileLabel.textContent = file.name; + window._quickClientItemId = crypto.randomUUID(); + window._quickSaved = null; } }); fileInput.onchange = e => { currentFile = e.target.files[0]; if (currentFile) { fileLabel.textContent = currentFile.name; + window._quickClientItemId = crypto.randomUUID(); + window._quickSaved = null; } }; + ['input', 'paste'].forEach(eventName => { + pasteText.addEventListener(eventName, () => { + window._quickClientItemId = crypto.randomUUID(); + window._quickSaved = null; + }); + }); importForm.onsubmit = async e => { e.preventDefault(); @@ -1671,6 +1729,10 @@ function setupQuickImportForm() { } else { fd.append('text', pasteText.value); } + fd.append( + 'client_item_id', + window._quickClientItemId || (window._quickClientItemId = crypto.randomUUID()) + ); const facet = quickFacetSelect ? quickFacetSelect.value : ''; if (facet) { @@ -1703,7 +1765,7 @@ function setupQuickImportForm() { }; } -async function uploadGuidedSourceFile() { +async function uploadGuidedSourceFile(source) { if (!currentGuidedFile) { throw new Error('Please select a file to import.'); } @@ -1712,6 +1774,13 @@ async function uploadGuidedSourceFile() { const guidedSettingInput = document.getElementById('guidedSettingInput'); const fd = new FormData(); fd.append('file', currentGuidedFile); + fd.append( + 'client_item_id', + window._guidedClientItemId || (window._guidedClientItemId = crypto.randomUUID()) + ); + if (source && source.name !== 'recording') { + fd.append('source_hint', source.name); + } if (guidedFacetSelect && guidedFacetSelect.value) { fd.append('facet', guidedFacetSelect.value); } @@ -1730,27 +1799,21 @@ async function uploadGuidedSourceFile() { currentGuidedSaved = data; - // Show dedup notice if the file was previously imported - if (data.dedup) { - showDedupNotice(data.dedup); + if (isTerminalDuplicate(data)) { + showDedupNotice(data.duplicate); } return data; } -function showDedupNotice(dedup) { +function showDedupNotice(duplicate) { const container = document.getElementById('guidedDedupNotice'); if (!container) return; container.innerHTML = `
- this file was already imported on ${escapeHtml(dedup.imported_at)}${dedup.entry_count ? ` (${dedup.entry_count} entries)` : ''}. - + ${escapeHtml(duplicateNoticeText(duplicate))}
`; - document.getElementById('dedupForceBtn').addEventListener('click', () => { - window._guidedForceImport = true; - container.innerHTML = '
will re-import. click start import to proceed.
'; - }); } async function saveGuidedPath(source) { @@ -1763,11 +1826,21 @@ async function saveGuidedPath(source) { const guidedSettingInput = document.getElementById('guidedSettingInput'); const facet = guidedFacetSelect ? guidedFacetSelect.value : ''; const setting = guidedSettingInput ? guidedSettingInput.value.trim() : ''; + const payload = { + path: pathInput.value.trim(), + facet, + setting, + client_item_id: window._guidedClientItemId + || (window._guidedClientItemId = crypto.randomUUID()), + }; + if (source && source.name !== 'recording') { + payload.source_hint = source.name; + } const response = await fetch('/app/import/api/save-path', { method: 'POST', headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ path: pathInput.value.trim(), facet, setting }), + body: JSON.stringify(payload), }); const data = await response.json(); if (!response.ok) { @@ -1775,6 +1848,9 @@ async function saveGuidedPath(source) { } currentGuidedSaved = data; + if (isTerminalDuplicate(data)) { + showDedupNotice(data.duplicate); + } return data; } @@ -1792,15 +1868,18 @@ async function startGuidedImport(source) { if (source.input_type === 'path_input') { saved = currentGuidedSaved || await saveGuidedPath(source); } else { - saved = currentGuidedSaved || await uploadGuidedSourceFile(); + saved = currentGuidedSaved || await uploadGuidedSourceFile(source); + } + if (isTerminalDuplicate(saved)) { + showDedupNotice(saved.duplicate); + return; } const path = saved.path; const ts = saved.timestamp; const facet = guidedFacetSelect ? guidedFacetSelect.value : ''; const setting = guidedSettingInput ? guidedSettingInput.value.trim() : ''; - const force = !!(window._guidedForceImport); - const metaResponse = await fetch('/app/import/api/facet', { + const metaResponse = await fetch('/app/import/api/meta', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ path, facet, setting }), @@ -1814,7 +1893,7 @@ async function startGuidedImport(source) { const startResponse = await fetch('/app/import/api/start', { method: 'POST', headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ path, timestamp: ts, source: source.name, force }), + body: JSON.stringify({ path, timestamp: ts }), }); if (!startResponse.ok) { @@ -1830,7 +1909,6 @@ async function startGuidedImport(source) { source_display: source.display_name, }; trackPendingImport(ts); - window._guidedForceImport = false; loadImports(); navigateTo(`progress/${ts}`); } catch (err) { @@ -1925,13 +2003,17 @@ startBtn.addEventListener('click', async () => { if (window.showError) showError('Timestamp is required before starting the import.'); return; } + if (isTerminalDuplicate(window._quickSaved)) { + showDetect(window._quickSaved); + return; + } const originalText = startBtn.textContent; startBtn.disabled = true; startBtn.innerHTML = 'starting'; try { - const response = await fetch('/app/import/api/facet', { + const response = await fetch('/app/import/api/meta', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ path, facet, setting }), @@ -1945,7 +2027,7 @@ startBtn.addEventListener('click', async () => { const startResponse = await fetch('/app/import/api/start', { method: 'POST', headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ path, timestamp: ts, source: 'quick', force: false }), + body: JSON.stringify({ path, timestamp: ts }), }); if (!startResponse.ok) { diff --git a/solstone/convey/reasons.py b/solstone/convey/reasons.py index 222e66d69..2e3c866b7 100644 --- a/solstone/convey/reasons.py +++ b/solstone/convey/reasons.py @@ -297,6 +297,11 @@ IMPORT_CONFLICT = Reason( "I couldn't start that import because it already exists.", 409, ) +IMPORT_CLIENT_ID_CONFLICT = Reason( + "import_client_id_conflict", + "That client_item_id is already staged for different content.", + 409, +) IMPORT_METADATA_FAILED = Reason( "import_metadata_failed", "I couldn't update that import metadata.", diff --git a/solstone/think/import_client.py b/solstone/think/import_client.py index 6e0a94de8..31a6c12e3 100644 --- a/solstone/think/import_client.py +++ b/solstone/think/import_client.py @@ -9,6 +9,7 @@ import argparse import json import logging import sys +import uuid from pathlib import Path from typing import Any @@ -57,7 +58,10 @@ def _build_parser() -> argparse.ArgumentParser: ) parser.add_argument( "--source", - help="Import source type (apple, plaud, audio, text, or a file importer name)", + help=( + "Import source hint: a file importer name (e.g. obsidian, plaud, ics) " + "or a canonical category (audio, image, document, text)" + ), ) parser.add_argument( "--force", @@ -165,13 +169,19 @@ def _payload_value(value: str | None) -> str | None: return stripped or None -def _save_media(client: ConveyClient, args: argparse.Namespace) -> dict[str, Any]: +def _save_media( + client: ConveyClient, + args: argparse.Namespace, + client_item_id: str, +) -> dict[str, Any]: media_path = Path(args.media).expanduser() data = { key: value for key, value in { + "client_item_id": client_item_id, "facet": _payload_value(args.facet), "setting": _payload_value(args.setting), + "source_hint": _payload_value(args.source), }.items() if value is not None } @@ -216,13 +226,6 @@ def _start_import( "timestamp": timestamp, "force": bool(args.force), } - for key, value in { - "facet": _payload_value(args.facet), - "setting": _payload_value(args.setting), - "source": _payload_value(args.source), - }.items(): - if value is not None: - payload[key] = value start_response = client.request("POST", f"{IMPORT_API}/start", json=payload) if not isinstance(start_response, dict): raise ConveyClientError(MALFORMED_RESPONSE) @@ -294,9 +297,35 @@ def _print_success( print("queued processing") +def _print_duplicate(save_response: dict[str, Any], *, json_out: bool) -> None: + if json_out: + print(json.dumps(save_response, sort_keys=True)) + return + + duplicate = save_response.get("duplicate") + if not isinstance(duplicate, dict): + print("sol import: duplicate import; skipping") + return + + state = duplicate.get("state") + import_id = duplicate.get("import_id") or "unknown" + if state == "imported": + imported_at = duplicate.get("imported_at") or "unknown date" + entry_count = duplicate.get("entry_count") + entries = f" ({entry_count} entries)" if entry_count is not None else "" + print(f"sol import: already imported on {imported_at}{entries}; skipping") + return + if state == "staged": + print(f"sol import: already staged as {import_id}; skipping") + return + + print("sol import: duplicate import; skipping") + + def _run(args: argparse.Namespace, client: ConveyClient) -> int: + client_item_id = uuid.uuid4().hex try: - save_response = _save_media(client, args) + save_response = _save_media(client, args, client_item_id) if not isinstance(save_response, dict): raise ConveyClientError(MALFORMED_RESPONSE) except ConveyUnreachableError: @@ -309,6 +338,13 @@ def _run(args: argparse.Namespace, client: ConveyClient) -> int: _print_client_error("stage import", err) return 1 + if ( + save_response.get("status") == "duplicate" + or save_response.get("recommended_action") == "do_not_start" + ): + _print_duplicate(save_response, json_out=bool(args.json)) + return 0 + staged_path = str(save_response.get("path") or args.media) try: start_response = _start_import(client, args, save_response) diff --git a/solstone/think/importers/cli.py b/solstone/think/importers/cli.py index ffc66e0c5..d4fec19c0 100644 --- a/solstone/think/importers/cli.py +++ b/solstone/think/importers/cli.py @@ -403,8 +403,6 @@ def _import_one_from_args(args: argparse.Namespace) -> dict[str, Any] | None: global _stage_start_time, _stages_run, _status_thread, _status_running args.media = os.path.expanduser(args.media) - if args.source == "quick": - args.source = None _file_importer = None import_source = None @@ -434,8 +432,10 @@ def _import_one_from_args(args: argparse.Namespace) -> dict[str, Any] | None: _file_importer = detected import_source = detected.name - # Try file importer detection for unknown file extensions - if _file_importer is None and not args.source: + # Try file importer detection for unknown file extensions. Non-file-importer + # --source values are intentionally ignored, so they should not suppress + # detection for image/structured imports. + if _file_importer is None: _ext = os.path.splitext(args.media)[1].lower() if _ext not in {".m4a", ".txt", ".md", ".pdf"}: from solstone.think.importers.file_importer import detect_file_importer @@ -537,17 +537,13 @@ def _import_one_from_args(args: argparse.Namespace) -> dict[str, Any] | None: day = base_dt.strftime("%Y%m%d") # --- Derive import_source for non-file-importer paths --- + # Generic (non-file-importer) items stream as audio or text by extension. + # A non-importer --source value (e.g. legacy "recording"/"quick") never + # becomes a stream name; it is ignored here so leaked vocabulary cannot + # reach the stream/manifest. File-importer sources are handled earlier. if import_source is None: - if args.source: - import_source = args.source - else: - _ext = os.path.splitext(args.media)[1].lower() - if _ext == ".m4a": - import_source = "apple" - elif _ext in {".txt", ".md", ".pdf"}: - import_source = "text" - else: - import_source = "audio" + _ext = os.path.splitext(args.media)[1].lower() + import_source = "text" if _ext in {".txt", ".md", ".pdf"} else "audio" stream = stream_name(import_source=import_source) needs_setup = _file_importer is None and not _is_in_imports(args.media) @@ -1347,7 +1343,11 @@ def main() -> None: "--source", type=str, default=None, - help="Import source type (apple, plaud, audio, text, or a file importer name). Auto-detected if omitted.", + help=( + "Import source hint: a file importer name (e.g. obsidian, plaud, ics) " + "or a canonical category (audio, image, document, text). " + "Auto-detected if omitted." + ), ) parser.add_argument( "--force", diff --git a/solstone/think/importers/utils.py b/solstone/think/importers/utils.py index 51faed999..58cdcc8a0 100644 --- a/solstone/think/importers/utils.py +++ b/solstone/think/importers/utils.py @@ -161,6 +161,49 @@ def read_import_metadata( raise FileNotFoundError(f"Import metadata not found for {timestamp}") +def find_staged_by_client_item_id( + journal_root: Path, client_item_id: str +) -> dict | None: + """Return staged import metadata with matching client_item_id, else None.""" + for timestamp in list_import_timestamps(journal_root): + try: + metadata = read_import_metadata(journal_root, timestamp) + except (FileNotFoundError, json.JSONDecodeError): + continue + if not isinstance(metadata, dict): + continue + if metadata.get("client_item_id") == client_item_id: + result = dict(metadata) + result["timestamp"] = timestamp + return result + + return None + + +def find_staged_by_source_hash( + journal_root: Path, + source_hash: str, + *, + exclude_timestamp: str | None = None, +) -> dict | None: + """Return staged import metadata with matching source_hash, else None.""" + for timestamp in list_import_timestamps(journal_root): + if timestamp == exclude_timestamp: + continue + try: + metadata = read_import_metadata(journal_root, timestamp) + except (FileNotFoundError, json.JSONDecodeError): + continue + if not isinstance(metadata, dict): + continue + if metadata.get("source_hash") == source_hash: + result = dict(metadata) + result["timestamp"] = timestamp + return result + + return None + + def update_import_metadata_fields( journal_root: Path, timestamp: str, diff --git a/solstone/think/media.py b/solstone/think/media.py index 45be712c1..94e345372 100644 --- a/solstone/think/media.py +++ b/solstone/think/media.py @@ -3,6 +3,8 @@ """Media format registry - single source of truth for extensions, MIME types, and kind.""" +from pathlib import Path + FORMATS = [ (".flac", "audio/flac", "audio"), (".opus", "audio/opus", "audio"), @@ -27,3 +29,55 @@ IMAGE_EXTENSIONS: frozenset[str] = frozenset( {".png", ".jpg", ".jpeg", ".heic", ".heif", ".gif", ".webp", ".tiff"} ) PDF_EXTENSIONS: frozenset[str] = frozenset({".pdf"}) + + +def canonical_source( + *, filename: str | None = None, content_type: str | None = None +) -> str: + """Infer the canonical import source category from content signals. + + Returns one of "audio", "image", "document", "text". This is a protocol + metadata category describing WHAT the content is; it does not drive importer + routing. + """ + suffix = Path(filename).suffix.lower() if filename else "" + if suffix in AUDIO_EXTENSIONS or suffix in VIDEO_EXTENSIONS: + return "audio" + if suffix in IMAGE_EXTENSIONS: + return "image" + if suffix in PDF_EXTENSIONS: + return "document" + + media_type = content_type.lower() if content_type else "" + if media_type.startswith(("audio/", "video/")): + return "audio" + if media_type.startswith("image/"): + return "image" + if media_type == "application/pdf": + return "document" + + return "text" + + +def canonical_source_signal( + *, filename: str | None = None, content_type: str | None = None +) -> str: + """Return the signal that determined the canonical source.""" + suffix = Path(filename).suffix.lower() if filename else "" + if ( + suffix in AUDIO_EXTENSIONS + or suffix in VIDEO_EXTENSIONS + or suffix in IMAGE_EXTENSIONS + or suffix in PDF_EXTENSIONS + ): + return "extension" + + media_type = content_type.lower() if content_type else "" + if ( + media_type.startswith(("audio/", "video/")) + or media_type.startswith("image/") + or media_type == "application/pdf" + ): + return "content_type" + + return "default" diff --git a/tests/test_import_client.py b/tests/test_import_client.py index 521bfdb2b..ff6bddfcb 100644 --- a/tests/test_import_client.py +++ b/tests/test_import_client.py @@ -57,6 +57,13 @@ class FakeClient: return self.request_responses[index] +def _pop_client_item_id(payload: dict[str, Any]) -> str: + client_item_id = payload.pop("client_item_id") + assert isinstance(client_item_id, str) + assert client_item_id + return client_item_id + + def test_mode_disposition_table_covers_d5_modes() -> None: assert import_client.MODE_DISPOSITIONS == { "positional_media": "http-client", @@ -89,15 +96,15 @@ def test_file_save_then_start( code = import_client.main([str(media)], client=client) # type: ignore[arg-type] assert code == 0 - assert client.uploads == [ - { - "path": "/app/import/api/save", - "files": { - "file": ("sample.txt", media, "application/octet-stream"), - }, - "data": {}, - } - ] + save_data = dict(client.uploads[0]["data"]) + client_item_id = _pop_client_item_id(save_data) + assert len(client_item_id) == 32 + assert len(client.uploads) == 1 + assert client.uploads[0]["path"] == "/app/import/api/save" + assert client.uploads[0]["files"] == { + "file": ("sample.txt", media, "application/octet-stream"), + } + assert save_data == {} assert client.requests == [ { "method": "POST", @@ -130,12 +137,13 @@ def test_save_path_then_start(tmp_path: Path) -> None: assert code == 0 assert client.uploads == [] - assert client.requests[0] == { - "method": "POST", - "path": "/app/import/api/save-path", - "params": None, - "json": {"path": str(media_dir)}, - } + save_json = dict(client.requests[0]["json"]) + client_item_id = _pop_client_item_id(save_json) + assert len(client_item_id) == 32 + assert client.requests[0]["method"] == "POST" + assert client.requests[0]["path"] == "/app/import/api/save-path" + assert client.requests[0]["params"] is None + assert save_json == {"path": str(media_dir)} assert client.requests[1]["path"] == "/app/import/api/start" assert client.requests[1]["json"]["timestamp"] == "20260101_130000" @@ -175,14 +183,17 @@ def test_metadata_and_start_options_forward(tmp_path: Path) -> None: ) assert code == 0 - assert client.uploads[0]["data"] == {"facet": "work", "setting": "office"} + save_data = dict(client.uploads[0]["data"]) + _pop_client_item_id(save_data) + assert save_data == { + "facet": "work", + "setting": "office", + "source_hint": "ics", + } assert client.requests[0]["json"] == { "path": "/journal/imports/20260101_120000/sample.txt", "timestamp": "20260101_120000", "force": True, - "facet": "work", - "setting": "office", - "source": "ics", } @@ -197,7 +208,9 @@ def test_deterministic_only_forwards_only_on_save_data(tmp_path: Path) -> None: ) assert code == 0 - assert client.uploads[0]["data"] == {"deterministic_only": "true"} + save_data = dict(client.uploads[0]["data"]) + _pop_client_item_id(save_data) + assert save_data == {"deterministic_only": "true"} assert "deterministic_only" not in client.requests[0]["json"] @@ -206,7 +219,10 @@ def test_json_output_shape(tmp_path: Path, capsys: pytest.CaptureFixture[str]) - media.write_text("hello", encoding="utf-8") client = FakeClient() - code = import_client.main([str(media), "--json"], client=client) # type: ignore[arg-type] + code = import_client.main( + [str(media), "--json"], + client=client, # type: ignore[arg-type] + ) assert code == 0 payload = json.loads(capsys.readouterr().out) @@ -222,6 +238,68 @@ def test_json_output_shape(tmp_path: Path, capsys: pytest.CaptureFixture[str]) - } +def test_duplicate_save_skips_start_and_returns_success( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + media = tmp_path / "sample.txt" + media.write_text("hello", encoding="utf-8") + client = FakeClient( + upload_response={ + "schema_version": 1, + "status": "duplicate", + "recommended_action": "do_not_start", + "path": "/journal/imports/20260101_120000/sample.txt", + "timestamp": "20260101_120000", + "duplicate": { + "import_id": "20260101_120000", + "imported_at": "2026-01-01T12:00:00", + "entry_count": 2, + "state": "imported", + }, + } + ) + + code = import_client.main([str(media)], client=client) # type: ignore[arg-type] + + assert code == 0 + assert client.requests == [] + captured = capsys.readouterr() + assert ( + "sol import: already imported on 2026-01-01T12:00:00 (2 entries); skipping" + ) in captured.out + assert captured.err == "" + + +def test_duplicate_json_outputs_save_response_without_start( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + media = tmp_path / "sample.txt" + media.write_text("hello", encoding="utf-8") + duplicate_response = { + "schema_version": 1, + "status": "duplicate", + "recommended_action": "do_not_start", + "path": "/journal/imports/20260101_120000/sample.txt", + "timestamp": "20260101_120000", + "duplicate": { + "import_id": "20260101_120000", + "imported_at": None, + "entry_count": None, + "state": "staged", + }, + } + client = FakeClient(upload_response=duplicate_response) + + code = import_client.main( + [str(media), "--json"], + client=client, # type: ignore[arg-type] + ) + + assert code == 0 + assert client.requests == [] + assert json.loads(capsys.readouterr().out) == duplicate_response + + def test_unreachable_is_clean( tmp_path: Path, capsys: pytest.CaptureFixture[str] ) -> None: diff --git a/tests/test_import_start.py b/tests/test_import_start.py index 374052bef..b1ffb22e7 100644 --- a/tests/test_import_start.py +++ b/tests/test_import_start.py @@ -3,18 +3,31 @@ from __future__ import annotations +import hashlib +import io import json -from importlib import import_module from pathlib import Path +from types import SimpleNamespace import pytest -from flask import Flask +from flask import Flask, g import solstone.convey.state as convey_state import solstone.think.utils as think_utils -from solstone.convey.reasons import IMPORT_CONFLICT, IMPORT_NOT_FOUND +from solstone.convey.reasons import ( + IMPORT_CLIENT_ID_CONFLICT, + IMPORT_CONFLICT, + IMPORT_NOT_FOUND, + INVALID_OPERATION_FOR_STATE, + MISSING_REQUIRED_FIELD, +) +from solstone.think.importers.utils import ( + read_import_metadata, + update_import_metadata_fields, + write_import_metadata, +) -import_routes = import_module("solstone.apps.import.routes") +import_routes = __import__("solstone.apps.import.routes", fromlist=["routes"]) @pytest.fixture @@ -26,14 +39,366 @@ def journal_env(tmp_path, monkeypatch) -> Path: return tmp_path +@pytest.fixture(autouse=True) +def _stable_timestamp_detection(monkeypatch): + monkeypatch.setattr( + import_routes, + "resolve_created_deterministic", + lambda *args, **kwargs: None, + ) + + def _no_model(*args, **kwargs): + raise AssertionError("model timestamp detection should not run") + + monkeypatch.setattr(import_routes, "detect_created", _no_model) + + @pytest.fixture def client(journal_env): app = Flask(__name__) + + @app.before_request + def _identity() -> None: + g.identity = SimpleNamespace(mode="local", fingerprint=None) + app.register_blueprint(import_routes.import_bp) return app.test_client() -def test_import_start_moves_staging_dir_and_updates_file_path( +def _sha(content: bytes) -> str: + return hashlib.sha256(content).hexdigest() + + +def _import_dirs(journal_root: Path) -> list[Path]: + return sorted( + path for path in (journal_root / "imports").iterdir() if path.is_dir() + ) + + +def _save_upload( + client, + *, + client_item_id: str, + content: bytes = b"sample audio", + filename: str = "sample.m4a", + content_type: str = "audio/mp4", + extra: dict | None = None, +): + data = { + "client_item_id": client_item_id, + "deterministic_only": "true", + "file": (io.BytesIO(content), filename, content_type), + } + if extra: + data.update(extra) + return client.post( + "/app/import/api/save", + data=data, + content_type="multipart/form-data", + ) + + +def _write_manifest( + journal_root: Path, + *, + import_id: str, + source_hash: str, + entry_count: int = 2, +) -> None: + import_dir = journal_root / "imports" / import_id + import_dir.mkdir(parents=True) + (import_dir / "manifest.json").write_text( + json.dumps( + { + "import_id": import_id, + "source_type": "audio", + "source_hash": source_hash, + "entry_count": entry_count, + "imported_at": "2026-01-01T12:00:00", + "imported_via": "cli", + } + ), + encoding="utf-8", + ) + + +def _write_staged_import( + journal_root: Path, + timestamp: str, + metadata: dict, +) -> Path: + import_dir = journal_root / "imports" / timestamp + import_dir.mkdir(parents=True) + media_path = import_dir / metadata.get("original_filename", "sample.m4a") + media_path.write_bytes(b"sample") + metadata = { + "file_path": str(media_path), + "user_timestamp": timestamp, + "source": "audio", + "source_inference": "extension", + "client": {}, + "facet": None, + "setting": None, + "imported_via": "web_dashboard", + "observer_handle": None, + "source_hint": None, + "mime_type": "audio/mp4", + **metadata, + } + write_import_metadata(journal_root, timestamp, metadata) + return media_path + + +def test_import_save_audio_upload_stages_versioned_summary(client, journal_env): + response = _save_upload( + client, + client_item_id="ios-audio-1", + content=b"audio bytes", + extra={"facet": "work", "client": json.dumps({"device": "ios"})}, + ) + + assert response.status_code == 200 + body = response.get_json() + assert body["schema_version"] == 1 + assert body["status"] == "staged" + assert body["replay"] is False + assert body["source"] == "audio" + assert body["client_item_id"] == "ios-audio-1" + assert body["recommended_action"] == "start" + assert body["facet"] == "work" + assert body["metadata"] == { + "original_filename": "sample.m4a", + "mime_type": "audio/mp4", + "imported_via": "web_dashboard", + "observer_handle": None, + "source_hint": None, + "client": {"device": "ios"}, + } + assert body["diagnostics"]["source_inference"] == "extension" + assert "timestamp_detection_method" not in body + assert "dedup" not in body + + metadata = read_import_metadata(journal_env, body["timestamp"]) + assert metadata["client_item_id"] == "ios-audio-1" + assert metadata["source_hash"] == _sha(b"audio bytes") + assert metadata["source"] == "audio" + assert metadata["client"] == {"device": "ios"} + + +def test_import_save_missing_client_item_id_returns_missing_required(client): + response = client.post( + "/app/import/api/save", + data={ + "deterministic_only": "true", + "file": (io.BytesIO(b"audio"), "sample.m4a", "audio/mp4"), + }, + content_type="multipart/form-data", + ) + + assert response.status_code == MISSING_REQUIRED_FIELD.status + assert response.get_json()["reason_code"] == MISSING_REQUIRED_FIELD.code + + +def test_import_save_replay_same_client_and_content_does_not_stage_again( + client, journal_env +): + first = _save_upload( + client, + client_item_id="ios-replay", + content=b"same bytes", + ) + before_dirs = _import_dirs(journal_env) + + second = _save_upload( + client, + client_item_id="ios-replay", + content=b"same bytes", + ) + + assert second.status_code == 200 + body = second.get_json() + assert body["status"] == "staged" + assert body["replay"] is True + assert body["recommended_action"] == "start" + assert body["path"] == first.get_json()["path"] + assert _import_dirs(journal_env) == before_dirs + + +def test_import_save_replay_started_item_does_not_recommend_start(client, journal_env): + first = _save_upload( + client, + client_item_id="ios-replay-started", + content=b"same started bytes", + ) + timestamp = first.get_json()["timestamp"] + update_import_metadata_fields( + journal_root=journal_env, + timestamp=timestamp, + updates={"task_id": "task-started"}, + ) + before_dirs = _import_dirs(journal_env) + + second = _save_upload( + client, + client_item_id="ios-replay-started", + content=b"same started bytes", + ) + + assert second.status_code == 200 + body = second.get_json() + assert body["status"] == "staged" + assert body["replay"] is True + assert body["recommended_action"] == "do_not_start" + assert body["path"] == first.get_json()["path"] + assert _import_dirs(journal_env) == before_dirs + + +def test_import_save_same_client_different_content_conflicts(client): + _save_upload(client, client_item_id="ios-conflict", content=b"one") + + response = _save_upload(client, client_item_id="ios-conflict", content=b"two") + + assert response.status_code == IMPORT_CLIENT_ID_CONFLICT.status + assert response.get_json()["reason_code"] == IMPORT_CLIENT_ID_CONFLICT.code + + +def test_import_save_duplicate_imported_content_is_terminal(client, journal_env): + content = b"already imported" + _write_manifest( + journal_env, + import_id="20260101_120000", + source_hash=_sha(content), + entry_count=3, + ) + before_dirs = _import_dirs(journal_env) + + response = _save_upload( + client, + client_item_id="ios-imported-dup", + content=content, + ) + + assert response.status_code == 200 + body = response.get_json() + assert body["status"] == "duplicate" + assert body["replay"] is False + assert body["recommended_action"] == "do_not_start" + assert body["duplicate"] == { + "import_id": "20260101_120000", + "imported_at": "2026-01-01T12:00:00", + "entry_count": 3, + "state": "imported", + } + assert _import_dirs(journal_env) == before_dirs + assert not any((path / "import.json").exists() for path in before_dirs) + + +def test_import_save_duplicate_staged_content_is_terminal(client, journal_env): + content = b"already staged" + _write_staged_import( + journal_env, + "20260101_121500", + { + "original_filename": "existing.m4a", + "client_item_id": "other-client", + "source_hash": _sha(content), + }, + ) + before_dirs = _import_dirs(journal_env) + + response = _save_upload( + client, + client_item_id="ios-staged-dup", + content=content, + ) + + assert response.status_code == 200 + body = response.get_json() + assert body["status"] == "duplicate" + assert body["client_item_id"] == "ios-staged-dup" + assert body["recommended_action"] == "do_not_start" + assert body["duplicate"] == { + "import_id": "20260101_121500", + "imported_at": None, + "entry_count": None, + "state": "staged", + } + assert _import_dirs(journal_env) == before_dirs + + +def test_import_meta_updates_facet_and_setting(client, journal_env): + media_path = _write_staged_import( + journal_env, + "20260101_130000", + {"original_filename": "meta.m4a", "client_item_id": "meta-client"}, + ) + + response = client.post( + "/app/import/api/meta", + json={"path": str(media_path), "facet": "work", "setting": "office"}, + ) + + assert response.status_code == 200 + assert response.get_json() == { + "status": "ok", + "path": str(media_path), + "timestamp": "20260101_130000", + "updated": {"facet": "work", "setting": "office"}, + } + metadata = read_import_metadata(journal_env, "20260101_130000") + assert metadata["facet"] == "work" + assert metadata["setting"] == "office" + + +def test_import_meta_missing_path_returns_missing_required(client): + response = client.post("/app/import/api/meta", json={}) + + assert response.status_code == MISSING_REQUIRED_FIELD.status + assert response.get_json()["reason_code"] == MISSING_REQUIRED_FIELD.code + + +def test_import_facet_route_removed_no_alias(client): + rules = list(client.application.url_map.iter_rules()) + + assert not any(rule.rule == "/app/import/api/facet" for rule in rules) + assert any( + rule.rule == "/app/import/api/meta" and "POST" in rule.methods for rule in rules + ) + + +def test_import_meta_missing_item_returns_import_not_found(client, journal_env): + missing_path = journal_env / "imports" / "20260101_130001" / "sample.m4a" + + response = client.post( + "/app/import/api/meta", + json={"path": str(missing_path), "facet": "work"}, + ) + + assert response.status_code == IMPORT_NOT_FOUND.status + assert response.get_json()["reason_code"] == IMPORT_NOT_FOUND.code + + +def test_import_meta_started_item_returns_invalid_operation(client, journal_env): + media_path = _write_staged_import( + journal_env, + "20260101_130002", + { + "original_filename": "started.m4a", + "client_item_id": "started-client", + "task_id": "task-1", + }, + ) + + response = client.post( + "/app/import/api/meta", + json={"path": str(media_path), "facet": "work"}, + ) + + assert response.status_code == INVALID_OPERATION_FOR_STATE.status + assert response.get_json()["reason_code"] == INVALID_OPERATION_FOR_STATE.code + + +def test_import_start_moves_dir_uses_saved_metadata_and_omits_generic_source( client, journal_env, monkeypatch ): emitted: list[dict[str, object]] = [] @@ -46,12 +411,16 @@ def test_import_start_moves_staging_dir_and_updates_file_path( ) old_ts = "20260101_120000" new_ts = "20260101_121500" - import_dir = journal_env / "imports" / old_ts - import_dir.mkdir() - media_path = import_dir / "sample.m4a" - media_path.write_bytes(b"sample") - (import_dir / "import.json").write_text( - json.dumps({"facet": "work"}), encoding="utf-8" + media_path = _write_staged_import( + journal_env, + old_ts, + { + "original_filename": "sample.m4a", + "client_item_id": "start-client", + "facet": "work", + "setting": "office", + "source_hint": None, + }, ) response = client.post( @@ -59,17 +428,17 @@ def test_import_start_moves_staging_dir_and_updates_file_path( json={ "path": str(media_path), "timestamp": new_ts, - "source": "audio", + "source": "recording", + "facet": "ignored", "force": True, }, ) assert response.status_code == 200 - assert response.get_json()["task_id"] new_dir = journal_env / "imports" / new_ts assert new_dir.exists() - assert not import_dir.exists() - metadata = json.loads((new_dir / "import.json").read_text(encoding="utf-8")) + assert not (journal_env / "imports" / old_ts).exists() + metadata = read_import_metadata(journal_env, new_ts) assert metadata["file_path"] == str(new_dir / media_path.name) assert emitted == [ { @@ -83,14 +452,89 @@ def test_import_start_moves_staging_dir_and_updates_file_path( new_ts, "--facet", "work", - "--source", - "audio", + "--setting", + "office", "--force", ], } ] +def test_import_start_forwards_only_saved_source_hint(client, journal_env, monkeypatch): + emitted: list[dict[str, object]] = [] + monkeypatch.setattr( + import_routes, + "emit", + lambda tract, event, **payload: emitted.append( + {"tract": tract, "event": event, **payload} + ), + ) + ts = "20260101_122000" + media_path = _write_staged_import( + journal_env, + ts, + { + "original_filename": "vault", + "client_item_id": "source-hint-client", + "source_hint": "obsidian", + }, + ) + + response = client.post( + "/app/import/api/start", + json={"path": str(media_path), "timestamp": ts, "source": "quick"}, + ) + + assert response.status_code == 200 + assert emitted[0]["cmd"] == [ + "journal", + "importer", + str(media_path), + ts, + "--source", + "obsidian", + ] + + +def test_import_start_refuses_terminal_duplicate_even_with_force( + client, journal_env, monkeypatch +): + emitted: list[dict[str, object]] = [] + monkeypatch.setattr( + import_routes, + "emit", + lambda tract, event, **payload: emitted.append( + {"tract": tract, "event": event, **payload} + ), + ) + content = b"terminal" + source_hash = _sha(content) + ts = "20260101_123000" + media_path = _write_staged_import( + journal_env, + ts, + { + "original_filename": "terminal.m4a", + "client_item_id": "terminal-client", + "source_hash": source_hash, + }, + ) + _write_manifest( + journal_env, + import_id="20260101_124500", + source_hash=source_hash, + ) + + response = client.post( + "/app/import/api/start", + json={"path": str(media_path), "timestamp": ts, "force": True}, + ) + + assert response.status_code == INVALID_OPERATION_FOR_STATE.status + assert response.get_json()["reason_code"] == INVALID_OPERATION_FOR_STATE.code + assert emitted == [] + + def test_import_start_missing_source_returns_import_not_found(client, journal_env): old_ts = "20260101_120000" new_ts = "20260101_121500" @@ -115,6 +559,15 @@ def test_import_start_target_exists_returns_import_conflict(client, journal_env) new_dir.mkdir() media_path = old_dir / "sample.m4a" media_path.write_bytes(b"sample") + write_import_metadata( + journal_env, + old_ts, + { + "file_path": str(media_path), + "user_timestamp": old_ts, + "client_item_id": "conflict-client", + }, + ) response = client.post( "/app/import/api/start", diff --git a/tests/test_importer.py b/tests/test_importer.py index cc7529e12..707a8dccb 100644 --- a/tests/test_importer.py +++ b/tests/test_importer.py @@ -8,6 +8,7 @@ import json import os import subprocess import time +import uuid import zipfile from io import BytesIO from pathlib import Path @@ -127,6 +128,7 @@ def _import_route_client( def _post_import_save(client, data: dict): payload = { "file": (BytesIO(b"hello"), "note.txt"), + "client_item_id": uuid.uuid4().hex, **data, } return client.post( @@ -319,9 +321,9 @@ def test_import_save_deterministic_success_skips_model_and_audits( assert response.status_code == 200 body = response.get_json() assert body["timestamp"] == "20240115_103000" - assert body["timestamp_detection_method"] == "deterministic" - assert body["timestamp_detection_model_called"] is False - assert body["timestamp_detection_no_match_reason"] is None + assert body["diagnostics"]["timestamp_detection_method"] == "deterministic" + assert body["diagnostics"]["timestamp_detection_model_called"] is False + assert body["diagnostics"]["timestamp_detection_no_match_reason"] is None metadata = _read_import_metadata(tmp_path, body["timestamp"]) assert metadata["detection_result"] == deterministic_result assert metadata["detected_timestamp"] == "20240115_103000" @@ -352,9 +354,12 @@ def test_import_save_deterministic_only_no_match_uses_upload_fallback_and_audit( "%Y%m%d_%H%M%S" ) assert body["timestamp"] == expected_timestamp - assert body["timestamp_detection_method"] == "upload_fallback" - assert body["timestamp_detection_model_called"] is False - assert body["timestamp_detection_no_match_reason"] == "no_deterministic_match" + assert body["diagnostics"]["timestamp_detection_method"] == "upload_fallback" + assert body["diagnostics"]["timestamp_detection_model_called"] is False + assert ( + body["diagnostics"]["timestamp_detection_no_match_reason"] + == "no_deterministic_match" + ) metadata = _read_import_metadata(tmp_path, body["timestamp"]) assert metadata["detected_timestamp"] is None assert metadata["user_timestamp"] == body["timestamp"] @@ -379,9 +384,9 @@ def test_import_save_model_success_audits(tmp_path, monkeypatch): assert response.status_code == 200 body = response.get_json() assert body["timestamp"] == "20240115_103000" - assert body["timestamp_detection_method"] == "model" - assert body["timestamp_detection_model_called"] is True - assert body["timestamp_detection_no_match_reason"] is None + assert body["diagnostics"]["timestamp_detection_method"] == "model" + assert body["diagnostics"]["timestamp_detection_model_called"] is True + assert body["diagnostics"]["timestamp_detection_no_match_reason"] is None metadata = _read_import_metadata(tmp_path, body["timestamp"]) assert metadata["detection_result"] == model_result assert metadata["timestamp_detection_method"] == "model" @@ -401,9 +406,11 @@ def test_import_save_model_no_match_audits_upload_fallback(tmp_path, monkeypatch assert response.status_code == 200 body = response.get_json() - assert body["timestamp_detection_method"] == "upload_fallback" - assert body["timestamp_detection_model_called"] is True - assert body["timestamp_detection_no_match_reason"] == "model_no_match" + assert body["diagnostics"]["timestamp_detection_method"] == "upload_fallback" + assert body["diagnostics"]["timestamp_detection_model_called"] is True + assert ( + body["diagnostics"]["timestamp_detection_no_match_reason"] == "model_no_match" + ) metadata = _read_import_metadata(tmp_path, body["timestamp"]) assert metadata["detected_timestamp"] is None assert metadata["timestamp_detection_method"] == "upload_fallback" diff --git a/tests/test_media.py b/tests/test_media.py new file mode 100644 index 000000000..4285d04d7 --- /dev/null +++ b/tests/test_media.py @@ -0,0 +1,66 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""Tests for media source inference helpers.""" + +import pytest + +from solstone.think.media import canonical_source, canonical_source_signal + + +@pytest.mark.parametrize( + ("filename", "expected"), + [ + ("recording.m4a", "audio"), + ("recording.mp3", "audio"), + ("recording.wav", "audio"), + ("clip.mp4", "audio"), + ("clip.mov", "audio"), + ("clip.webm", "audio"), + ("photo.png", "image"), + ("photo.jpg", "image"), + ("photo.heic", "image"), + ("doc.pdf", "document"), + ("note.txt", "text"), + ("note.md", "text"), + ("", "text"), + ], +) +def test_canonical_source_from_filename(filename, expected): + assert canonical_source(filename=filename) == expected + + +@pytest.mark.parametrize( + ("content_type", "expected"), + [ + ("audio/mp4", "audio"), + ("image/png", "image"), + ("application/pdf", "document"), + ("text/plain", "text"), + ], +) +def test_canonical_source_from_content_type(content_type, expected): + assert canonical_source(content_type=content_type) == expected + + +def test_canonical_source_prefers_extension_over_content_type(): + assert canonical_source(filename="doc.pdf", content_type="text/plain") == "document" + + +@pytest.mark.parametrize( + ("filename", "content_type", "expected"), + [ + ("recording.m4a", None, "extension"), + ("photo.png", "text/plain", "extension"), + (None, "audio/mp4", "content_type"), + (None, "image/png", "content_type"), + (None, "application/pdf", "content_type"), + ("note.txt", "text/plain", "default"), + (None, "text/plain", "default"), + ], +) +def test_canonical_source_signal(filename, content_type, expected): + assert ( + canonical_source_signal(filename=filename, content_type=content_type) + == expected + ) diff --git a/tests/test_openapi_contract.py b/tests/test_openapi_contract.py index ff39841d3..b22675185 100644 --- a/tests/test_openapi_contract.py +++ b/tests/test_openapi_contract.py @@ -48,7 +48,9 @@ CONTRACTED_PATHS = { "/api/voice/session", "/api/voice/status", "/app/home/api/pulse", + "/app/import/api/meta", "/app/import/api/save", + "/app/import/api/save-path", "/app/import/api/start", "/app/network/api/status", "/app/network/local-endpoints", @@ -79,7 +81,9 @@ CONTRACTED_INVENTORY_TRIPLES = { ("GET", "/api/voice/observer-actions", "voice.observerActions"), ("GET", "/api/voice/status", "voice.status"), ("GET", "/app/home/api/pulse", "home.pulse"), + ("POST", "/app/import/api/meta", "import.meta"), ("POST", "/app/import/api/save", "import.save"), + ("POST", "/app/import/api/save-path", "import.savePath"), ("POST", "/app/import/api/start", "import.start"), ("DELETE", "/app/observer/source/{stream}", "observer.deleteSource"), } @@ -338,7 +342,7 @@ def test_no_r0_routes_in_artifact(): assert "/api/config/convey" not in document["paths"] assert "/api/system/status" not in document["paths"] assert set(document["paths"]) == CONTRACTED_PATHS - assert len(document["paths"]) == 28 + assert len(document["paths"]) == 30 def test_home_pulse_named_fields_present(contract_app):