From 3415fd10145244e0d4853e18134e8ff191be2704 Mon Sep 17 00:00:00 2001 From: Drew Cain Date: Wed, 6 May 2026 12:21:35 -0500 Subject: [PATCH] fix(core): parse picoschema modifier descriptions (#796) Signed-off-by: Drew Cain Signed-off-by: phernandez Co-authored-by: phernandez --- src/basic_memory/schema/parser.py | 87 +++++++++++++---- tests/mcp/test_tool_schema.py | 65 +++++++++++++ tests/mcp/test_tool_search.py | 25 ++++- tests/schema/test_parser.py | 156 ++++++++++++++++++++++++++++++ 4 files changed, 309 insertions(+), 24 deletions(-) diff --git a/src/basic_memory/schema/parser.py b/src/basic_memory/schema/parser.py index 2d4a2a37..2d0f2685 100644 --- a/src/basic_memory/schema/parser.py +++ b/src/basic_memory/schema/parser.py @@ -8,7 +8,9 @@ Syntax reference: field: type, description # required field field?: type, description # optional field field(array): type # array of values + field(array, description): type # array with description field?(enum): [val1, val2] # enumeration + field?(enum, description): [val1, val2] # enum with description field?(object): # nested object sub_field: type EntityName as type (capitalized) # entity reference @@ -58,46 +60,88 @@ class SchemaDefinition: # with an uppercase letter is treated as an entity reference. SCALAR_TYPES = frozenset({"string", "integer", "number", "boolean", "any"}) +MODIFIER_TYPES = frozenset({"array", "enum", "object"}) # --- Field Name Parsing --- -def _parse_field_key(key: str) -> tuple[str, bool, bool, bool, bool]: +def _parse_field_key_parts(key: str) -> tuple[str, bool, bool, bool, bool, str | None]: """Parse a Picoschema field key into its components. - Returns (name, required, is_array, is_enum, is_object). - The key format is: name[?][(array|enum|object)] + Returns (name, required, is_array, is_enum, is_object, description). + The key format is: name[?][(array|enum|object[, description])] Examples: - "name" -> ("name", True, False, False) - "role?" -> ("role", False, False, False) - "tags?(array)" -> ("tags", False, True, False) - "status?(enum)" -> ("status", False, False, True) - "metadata?(object)" -> ("metadata", False, False, False) + children + "name" -> ("name", True, False, False, False, None) + "role?" -> ("role", False, False, False, False, None) + "tags?(array)" -> ("tags", False, True, False, False, None) + "tags?(array, labels)" -> ("tags", False, True, False, False, "labels") + "status?(enum)" -> ("status", False, False, True, False, None) + "metadata?(object)" -> ("metadata", False, False, False, True, None) """ required = True is_array = False is_enum = False is_object = False + description = None - # Check for modifier suffix: (array), (enum), (object) - if key.endswith("(array)"): + key, modifier, description = _split_modifier_suffix(key) + + if modifier == "array": is_array = True - key = key[: -len("(array)")] - elif key.endswith("(enum)"): + elif modifier == "enum": is_enum = True - key = key[: -len("(enum)")] - elif key.endswith("(object)"): + elif modifier == "object": is_object = True - key = key[: -len("(object)")] # Check for optional marker if key.endswith("?"): required = False key = key[:-1] - return key, required, is_array, is_enum, is_object + return key.strip(), required, is_array, is_enum, is_object, description + + +def _parse_field_key(key: str) -> tuple[str, bool, bool, bool, bool]: + """Parse a Picoschema field key, discarding any modifier description.""" + name, required, is_array, is_enum, is_object, _description = _parse_field_key_parts(key) + return name, required, is_array, is_enum, is_object + + +def _split_modifier_suffix(key: str) -> tuple[str, str | None, str | None]: + """Split a trailing picoschema modifier from a field key.""" + stripped_key = key.rstrip() + if not stripped_key.endswith(")"): + return key, None, None + + # Trigger: field names and modifier descriptions may both contain parentheses + # Why: only the parenthesis paired with the final suffix can introduce a modifier + # Outcome: preserves names like "risk(score)" and descriptions like "labels (freeform)" + open_paren_index = -1 + depth = 0 + for index in range(len(stripped_key) - 1, -1, -1): + char = stripped_key[index] + if char == ")": + depth += 1 + elif char == "(": + depth -= 1 + if depth == 0: + open_paren_index = index + break + + if open_paren_index == -1: + return key, None, None + + modifier_text = stripped_key[open_paren_index + 1 : -1].strip() + modifier, separator, description = modifier_text.partition(",") + modifier = modifier.strip() + if modifier not in MODIFIER_TYPES: + return key, None, None + + key_without_modifier = stripped_key[:open_paren_index].rstrip() + parsed_description = description.strip() if separator else None + return key_without_modifier, modifier, parsed_description or None def _parse_type_and_description(value: str) -> tuple[str, str | None]: @@ -170,7 +214,7 @@ def parse_picoschema(yaml_dict: dict) -> list[SchemaField]: fields: list[SchemaField] = [] for key, value in yaml_dict.items(): - name, required, is_array, is_enum, is_object = _parse_field_key(key) + name, required, is_array, is_enum, is_object, key_description = _parse_field_key_parts(key) # --- Enum fields --- # Trigger: value is a list or a string containing bracketed enum values @@ -179,11 +223,12 @@ def parse_picoschema(yaml_dict: dict) -> list[SchemaField]: # in YAML to avoid parse errors) # Outcome: SchemaField with is_enum=True and enum_values populated if is_enum: - description = None + description = key_description if isinstance(value, list): enum_values = [str(v) for v in value] else: - enum_values, description = _parse_enum_string(str(value)) + enum_values, value_description = _parse_enum_string(str(value)) + description = description or value_description fields.append( SchemaField( name=name, @@ -207,13 +252,15 @@ def parse_picoschema(yaml_dict: dict) -> list[SchemaField]: name=name, type="object", required=required, + description=key_description, children=children, ) ) continue # --- Scalar and entity ref fields --- - type_str, description = _parse_type_and_description(str(value)) + type_str, value_description = _parse_type_and_description(str(value)) + description = key_description or value_description is_entity_ref = _is_entity_ref_type(type_str) fields.append( diff --git a/tests/mcp/test_tool_schema.py b/tests/mcp/test_tool_schema.py index 50a5b198..8bc2c79e 100644 --- a/tests/mcp/test_tool_schema.py +++ b/tests/mcp/test_tool_schema.py @@ -116,6 +116,71 @@ async def test_schema_validate_json_output(app, test_project, sync_service): assert result["results"][0]["passed"] is True +@pytest.mark.asyncio +async def test_schema_validate_picoschema_modifier_descriptions(app, test_project, sync_service): + """Modifier descriptions should not become literal field names.""" + project_path = Path(test_project.path) + + _write_schema_file( + project_path, + "schemas/PicoTest.md", + """\ +--- +title: PicoTest +type: schema +entity: pico_test +schema: + name: string + status(enum, current state): [active, inactive] + tags(array, list of tags): string +settings: + validation: warn +--- + +# PicoTest +""", + ) + _write_schema_file( + project_path, + "pico/PicoTest1.md", + """\ +--- +title: PicoTest1 +type: pico_test +permalink: pico/pico-test-1 +--- + +# PicoTest1 + +## Observations +- [name] PicoTest1 +- [status] active +- [tags] foo +- [tags] bar +""", + ) + + await sync_service.sync(project_path) + + result = await schema_validate( + note_type="pico_test", + project=test_project.name, + output_format="json", + ) + + assert isinstance(result, dict) + note_result = result["results"][0] + field_statuses = {fr["field_name"]: fr["status"] for fr in note_result["field_results"]} + assert result["valid_count"] == 1 + assert note_result["warnings"] == [] + assert note_result["unmatched_observations"] == {} + assert field_statuses == { + "name": "present", + "status": "present", + "tags": "present", + } + + @pytest.mark.asyncio async def test_schema_validate_by_identifier(app, test_project, sync_service): """Validate a specific note by identifier.""" diff --git a/tests/mcp/test_tool_search.py b/tests/mcp/test_tool_search.py index 2582b167..69a690f1 100644 --- a/tests/mcp/test_tool_search.py +++ b/tests/mcp/test_tool_search.py @@ -28,7 +28,10 @@ async def test_search_text(client, test_project): # Search for it (use json format to inspect structured results) response = await search_notes( - project=test_project.name, query="searchable", output_format="json" + project=test_project.name, + query="searchable", + search_type="text", + output_format="json", ) # Verify results - handle both success and error cases @@ -183,7 +186,12 @@ async def test_search_pagination(client, test_project): # Search for it (use json format to inspect structured results) response = await search_notes( - project=test_project.name, query="searchable", page=1, page_size=1, output_format="json" + project=test_project.name, + query="searchable", + search_type="text", + page=1, + page_size=1, + output_format="json", ) # Verify results - handle both success and error cases @@ -212,7 +220,11 @@ async def test_search_with_type_filter(client, test_project): # Search with note type filter (use json format to inspect structured results) response = await search_notes( - project=test_project.name, query="type", note_types=["note"], output_format="json" + project=test_project.name, + query="type", + search_type="text", + note_types=["note"], + output_format="json", ) # Verify results - handle both success and error cases @@ -237,7 +249,11 @@ async def test_search_with_entity_type_filter(client, test_project): # Search with entity_types (SearchItemType) filter (use json format) response = await search_notes( - project=test_project.name, query="type", entity_types=["entity"], output_format="json" + project=test_project.name, + query="type", + search_type="text", + entity_types=["entity"], + output_format="json", ) # Verify results - handle both success and error cases @@ -265,6 +281,7 @@ async def test_search_with_date_filter(client, test_project): response = await search_notes( project=test_project.name, query="recent", + search_type="text", after_date=one_hour_ago.isoformat(), output_format="json", ) diff --git a/tests/schema/test_parser.py b/tests/schema/test_parser.py index e6876618..c688bc16 100644 --- a/tests/schema/test_parser.py +++ b/tests/schema/test_parser.py @@ -7,6 +7,7 @@ from basic_memory.schema.parser import ( parse_picoschema, parse_schema_note, _parse_field_key, + _parse_field_key_parts, _parse_type_and_description, _parse_enum_string, _is_entity_ref_type, @@ -43,18 +44,50 @@ class TestParseFieldKey: assert required is False assert is_array is True + def test_array_with_description_in_modifier(self): + name, required, is_array, is_enum, is_object = _parse_field_key("tags(array, list of tags)") + assert name == "tags" + assert required is True + assert is_array is True + assert is_enum is False + assert is_object is False + + def test_optional_array_with_description_in_modifier(self): + name, required, is_array, is_enum, is_object = _parse_field_key( + "tags?(array, list of tags)" + ) + assert name == "tags" + assert required is False + assert is_array is True + def test_enum(self): name, required, is_array, is_enum, is_object = _parse_field_key("status?(enum)") assert name == "status" assert required is False assert is_enum is True + def test_enum_with_description_in_modifier(self): + name, required, is_array, is_enum, is_object = _parse_field_key( + "status(enum, current state)" + ) + assert name == "status" + assert required is True + assert is_enum is True + def test_object(self): name, required, is_array, is_enum, is_object = _parse_field_key("metadata?(object)") assert name == "metadata" assert required is False assert is_object is True + def test_object_with_description_in_modifier(self): + name, required, is_array, is_enum, is_object = _parse_field_key( + "metadata?(object, nested metadata)" + ) + assert name == "metadata" + assert required is False + assert is_object is True + def test_required_enum(self): name, required, is_array, is_enum, is_object = _parse_field_key("status(enum)") assert name == "status" @@ -62,6 +95,68 @@ class TestParseFieldKey: assert is_enum is True +class TestParseFieldKeyParts: + def test_modifier_description_returned(self): + assert _parse_field_key_parts("tags(array, list of tags)") == ( + "tags", + True, + True, + False, + False, + "list of tags", + ) + + def test_optional_enum_description_returned(self): + assert _parse_field_key_parts("status?(enum, current state)") == ( + "status", + False, + False, + True, + False, + "current state", + ) + + def test_no_modifier_description_is_none(self): + assert _parse_field_key_parts("name") == ( + "name", + True, + False, + False, + False, + None, + ) + + def test_description_can_contain_parentheses(self): + assert _parse_field_key_parts("tags(array, labels (freeform))") == ( + "tags", + True, + True, + False, + False, + "labels (freeform)", + ) + + def test_field_name_can_contain_parentheses_before_modifier(self): + assert _parse_field_key_parts("risk(score)(array)") == ( + "risk(score)", + True, + True, + False, + False, + None, + ) + + def test_optional_field_name_can_contain_parentheses_before_modifier(self): + assert _parse_field_key_parts("risk(score)?(array, score buckets)") == ( + "risk(score)", + False, + True, + False, + False, + "score buckets", + ) + + # --- _parse_type_and_description --- @@ -152,6 +247,19 @@ class TestParsePicoschema: assert fields[0].is_array is True assert fields[0].required is False + def test_array_field_with_description_in_modifier(self): + fields = parse_picoschema({"tags(array, list of tags)": "string"}) + assert fields[0].name == "tags" + assert fields[0].type == "string" + assert fields[0].is_array is True + assert fields[0].description == "list of tags" + + def test_parenthesized_field_name_with_array_modifier(self): + fields = parse_picoschema({"risk(score)(array)": "string"}) + assert fields[0].name == "risk(score)" + assert fields[0].type == "string" + assert fields[0].is_array is True + def test_entity_ref_field(self): fields = parse_picoschema({"works_at?": "Organization, employer"}) assert fields[0].name == "works_at" @@ -165,6 +273,15 @@ class TestParsePicoschema: assert fields[0].is_enum is True assert fields[0].enum_values == ["active", "inactive"] + def test_enum_field_with_description_in_modifier(self): + fields = parse_picoschema({"status(enum, current state)": ["active", "inactive"]}) + assert fields[0].name == "status" + assert fields[0].type == "enum" + assert fields[0].required is True + assert fields[0].is_enum is True + assert fields[0].enum_values == ["active", "inactive"] + assert fields[0].description == "current state" + def test_enum_field_with_string(self): fields = parse_picoschema({"status?(enum)": "active"}) assert fields[0].is_enum is True @@ -204,6 +321,20 @@ class TestParsePicoschema: assert fields[0].children[0].name == "street" assert fields[0].children[1].name == "city" + def test_object_field_with_description_in_modifier(self): + fields = parse_picoschema( + { + "address?(object, mailing address)": { + "street": "string", + } + } + ) + assert fields[0].name == "address" + assert fields[0].type == "object" + assert fields[0].required is False + assert fields[0].description == "mailing address" + assert fields[0].children[0].name == "street" + def test_dict_value_treated_as_object(self): """A dict value without explicit (object) is still treated as an object.""" fields = parse_picoschema( @@ -315,6 +446,31 @@ class TestParseSchemaNote: assert status_field.is_enum is True assert status_field.enum_values == ["draft", "published"] + def test_settings_frontmatter_parses_modifier_descriptions(self): + frontmatter = { + "entity": "Person", + "schema": {"name": "string"}, + "settings": { + "validation": "warn", + "frontmatter": { + "tags?(array, note tags)": "string", + "status?(enum, publication state)": ["draft", "published"], + }, + }, + } + result = parse_schema_note(frontmatter) + + tags_field = next(f for f in result.frontmatter_fields if f.name == "tags") + assert tags_field.required is False + assert tags_field.is_array is True + assert tags_field.description == "note tags" + + status_field = next(f for f in result.frontmatter_fields if f.name == "status") + assert status_field.required is False + assert status_field.is_enum is True + assert status_field.enum_values == ["draft", "published"] + assert status_field.description == "publication state" + def test_no_settings_frontmatter_defaults_to_empty(self): frontmatter = { "entity": "Person",