fix(core): parse picoschema modifier descriptions (#796)

Signed-off-by: Drew Cain <groksrc@gmail.com>
Signed-off-by: phernandez <paul@basicmachines.co>
Co-authored-by: phernandez <paul@basicmachines.co>

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(

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."""

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",
     )

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",