fix(mcp): conform read-only observation tools to NL-DSL-LLM

project/ticket-078/README.md

@@ -0,0 +1,26 @@
+# Ticket 078: MCP observation contract
+
+- **ID**: ticket-078
+- **Owner**: codex-monag-mcp-20260919
+- **Status**: IN_PROGRESS
+- **Workflow state**: EDIT
+- **Created**: 2026-09-19
+- **Planfile**: PLF-028
+- **Issue**: https://github.com/semcod/monag/issues/95
+
+## Goal and scope
+
+SESSION_EXECUTION_AUTHORIZATION: user requested continuation after the installed MCP assessment, under the existing fix/publication authorization. Adopt the read-only observation interface from wellmanifest/nl-dsl-llm at 2040efe37b9eb898350f3fec0285a2d4e69d4e34. Implement execute_dsl, nl_ask, describe_grammar, schema://current, standard structured results and correct error propagation, with one validated observation executor including advisory queries. Scope is disjoint from ticket-076 and the concurrent governance adoption ticket-077. No claim of full mutation-interface conformance or client MCP registration.
+
+## Acceptance criteria
+
+- [x] AC-01: Standard MCP tools/resource and shared CLI/REST observation entrypoints preserve provenance, validate commands, reject malformed DSL and emit correct errors; legacy tool names remain usable.
+- [ ] AC-02: Regression/full suite and governance pass; independent protected publication and installed-runtime canaries succeed.
+
+## Standard source
+
+The immutable source, supported schema profile and invocation examples are returned by describe_grammar and schema://current; source and tests are the material deliverables.
+
+## Validation
+
+Final full application run: 362 tests and 30 subtests passed, including twelve conformance tests covering CLI compatibility and rejected LLM translation provenance. The pinned normative command/result schemas independently validate real success and error envelopes; the advertised observation schema is valid JSON Schema. Ruff on all changed Python files and governance pass. Protected review and installed-runtime validation remain external delivery receipts.

project/ticket-078/intent.json

@@ -0,0 +1,110 @@
+{
+  "schema": "new-project.intent/v3",
+  "ticket": "ticket-078",
+  "summary": "Conform MCP observation tools and shared CLI REST contract to NL-DSL-LLM",
+  "workstream": "application",
+  "classification": {
+    "kind": "BUG",
+    "priority": "P1",
+    "origin": "requested"
+  },
+  "allowedPaths": [
+    "project/ticket-078/**",
+    "src/monag/cli.py",
+    "src/monag/dsl.py",
+    "src/monag/dsl_llm.py",
+    "src/monag/mcp.py",
+    "src/monag/nl_contract.py",
+    "src/monag/panel.py",
+    "tests/test_nl_contract.py"
+  ],
+  "forbiddenPaths": [
+    "project/ticket-*/user-*.md"
+  ],
+  "stacks": [],
+  "dependsOn": [],
+  "conflictsWith": [],
+  "integrationTicket": null,
+  "delivery": {
+    "acceptedBaseSha": "fb5259da3fb739cb0d138bd9008d422b59232381",
+    "targetBranch": "main",
+    "outcome": "Expose standard MCP observation tools/resource, strict validated DSL execution and consistent structured results through MCP, CLI and REST; publish and deploy verified merged package",
+    "nonGoals": [
+      "No mutation command redesign or remote MCP registration",
+      "No edits to concurrent governance adoption or historical branches",
+      "No new runtime dependencies"
+    ],
+    "complexity": "L",
+    "estimatedMinutes": 75,
+    "budgets": {
+      "maxImplementationFiles": 9,
+      "maxAffectedComponents": 4,
+      "maxPublicInterfaceChanges": 3,
+      "maxRuntimeDependencies": 0
+    },
+    "architecture": {
+      "status": "accepted",
+      "decision": "Share a closed JSON-schema observation profile, its small validator, the existing DSL executor and standard result envelope across adapters; preserve legacy monag_* text alongside machine-readable results",
+      "components": [
+        {
+          "name": "observation-contract",
+          "paths": [
+            "src/monag/dsl.py",
+            "src/monag/dsl_llm.py",
+            "src/monag/nl_contract.py"
+          ]
+        },
+        {
+          "name": "mcp",
+          "paths": [
+            "src/monag/mcp.py"
+          ]
+        },
+        {
+          "name": "cli-rest",
+          "paths": [
+            "src/monag/cli.py",
+            "src/monag/panel.py"
+          ]
+        },
+        {
+          "name": "conformance",
+          "paths": [
+            "tests/test_nl_contract.py"
+          ]
+        }
+      ],
+      "responsibilityChanges": false,
+      "interfaceChanges": [
+        "MCP standard tools and schema resource",
+        "CLI dsl and structured ask",
+        "REST versioned observation routes"
+      ],
+      "dataChanges": [],
+      "ui": {
+        "impact": "none",
+        "states": [],
+        "evidence": []
+      },
+      "rollback": "Revert the protected merge; switch installed launcher back to its recorded previous runtime"
+    },
+    "runtimeDependencies": [],
+    "validation": [
+      {
+        "criterion": "AC-01",
+        "commands": [
+          "python -m pytest -q tests/test_nl_contract.py tests/test_mcp.py tests/test_dsl.py tests/test_dsl_llm.py"
+        ],
+        "evidence": "Strict invalid-input rejection, real stdio/HTTP/CLI parity, provider fallback and legacy tool compatibility"
+      },
+      {
+        "criterion": "AC-02",
+        "commands": [
+          "python -m pytest -q",
+          "./project/governance-check.sh --actor agent"
+        ],
+        "evidence": "Full tests and protected exact-head checks pass; installed merged MCP handshake/tool call verified"
+      }
+    ]
+  }
+}

src/monag/cli.py

@@ -277,6 +277,8 @@ def main(argv=None):
     query_parser = sub.add_parser('query', aliases=['ask'],
                                   help='execute a natural language query or OBSERVE DSL command')
     query_parser.add_argument('query', nargs='+', help='natural language query phrase or OBSERVE DSL command')
+    dsl_parser = sub.add_parser('dsl', help='execute one validated OBSERVE statement')
+    dsl_parser.add_argument('query', nargs='+', help='canonical OBSERVE DSL')
     sub.add_parser('catalog', help='read-only, local-only catalog of what each repository under --root '
                                    'declares itself to be (description, stack, entry points)')
     export_parser = sub.add_parser('export', help='read-only staging list of candidate work items '
@@ -594,7 +596,19 @@ def display_report(document):
             from . import mcp
             mcp.run_stdio_server(root, depth=args.depth)
             return 0
-        if args.mode in ('query', 'ask'):
+        if args.mode in ('ask', 'dsl'):
+            from . import nl_contract
+            result = nl_contract.execute(' '.join(args.query), root,
+                                         direct=args.mode == 'dsl', depth=args.depth,
+                                         registry=registry)
+            if output_format == 'json':
+                print(json.dumps(result, ensure_ascii=True))
+            elif result['success']:
+                display_report(result['meta']['markdown'])
+            else:
+                print(result['errors'][0]['message'], file=sys.stderr)
+            return 0 if result['success'] else 1
+        if args.mode == 'query':
             from . import dsl_llm
             query_str = ' '.join(args.query)
             res = dsl_llm.execute(query_str, root, depth=args.depth, registry=registry)
@@ -606,7 +620,7 @@ def display_report(document):
                 else:
                     print(f"Error: {res.get('error')}", file=sys.stderr)
                     return 1
-            return 0
+            return 0 if res.get('status') == 'ok' else 1
         if args.mode == 'catalog':
             from . import catalog
             if output_format != 'json' and sys.stderr.isatty():

src/monag/dsl.py

@@ -20,27 +20,34 @@
 
 SCHEMA = 'monag.dsl/v1'
 
-VALID_DOMAINS = {'prs', 'pr', 'audit', 'status', 'agents', 'resume', 'usage', 'catalog'}
+VALID_DOMAINS = {'prs', 'pr', 'audit', 'status', 'agents', 'resume', 'usage', 'catalog', 'advise'}
 
 
 class Query:
     """Structured representation of a Monag observation query."""
 
     def __init__(self, target, hours=24.0, state='all', limit=20, unpushed_only=False,
-                 worktrees_only=False, raw_input=None):
+                 worktrees_only=False, raw_input=None, issue_limit=None,
+                 radar=False, tier="all", emit_planfile=False):
         # Normalize target
         if target in {'pr', 'prs'}:
             self.target = 'prs'
         elif target in {'agents', 'status'}:
             self.target = 'status'
         else:
             self.target = target
-        self.hours = float(hours) if hours is not None else 24.0
-        self.state = state if state in {'open', 'merged', 'all'} else 'all'
-        self.limit = int(limit) if limit is not None else 20
-        self.unpushed_only = bool(unpushed_only)
-        self.worktrees_only = bool(worktrees_only)
+        self.hours = hours if hours is not None else 24.0
+        self.state = state
+        self.limit = limit if limit is not None else 20
+        self.unpushed_only = unpushed_only
+        self.worktrees_only = worktrees_only
+        self.issue_limit = issue_limit
+        self.radar = radar
+        self.tier = tier
+        self.emit_planfile = emit_planfile
         self.raw_input = raw_input or ''
+        from .nl_contract import validate_query
+        validate_query(self)
 
     def to_dsl(self):
         """Serialize query to canonical DSL string."""
@@ -55,6 +62,14 @@ def to_dsl(self):
             parts.append('UNPUSHED_ONLY')
         if self.worktrees_only:
             parts.append('WORKTREES_ONLY')
+        if self.issue_limit is not None:
+            parts.extend(['ISSUE_LIMIT', str(self.issue_limit)])
+        if self.radar:
+            parts.append('RADAR')
+        if self.tier != 'all':
+            parts.extend(['TIER', self.tier])
+        if self.emit_planfile:
+            parts.append('EMIT_PLANFILE')
         return ' '.join(parts)
 
     def to_dict(self):
@@ -65,6 +80,10 @@ def to_dict(self):
             'limit': self.limit,
             'unpushed_only': self.unpushed_only,
             'worktrees_only': self.worktrees_only,
+            'issue_limit': self.issue_limit,
+            'radar': self.radar,
+            'tier': self.tier,
+            'emit_planfile': self.emit_planfile,
             'dsl': self.to_dsl(),
             'raw_input': self.raw_input,
         }
@@ -97,34 +116,28 @@ def parse_dsl(text):
         return None
 
     kwargs = {'target': target, 'raw_input': text}
+    converters = {'HOURS': float, 'LIMIT': int, 'ISSUE_LIMIT': int,
+                  'STATE': str.lower, 'TIER': str.lower}
+    flags = {'UNPUSHED_ONLY', 'WORKTREES_ONLY', 'RADAR', 'EMIT_PLANFILE'}
+    seen = set()
     i = 2
-    while i < len(tokens):
-        key = tokens[i].upper()
-        if key == 'HOURS' and i + 1 < len(tokens):
-            try:
-                kwargs['hours'] = float(tokens[i + 1])
-            except ValueError:
-                pass
-            i += 2
-        elif key == 'STATE' and i + 1 < len(tokens):
-            kwargs['state'] = tokens[i + 1].lower()
-            i += 2
-        elif key == 'LIMIT' and i + 1 < len(tokens):
-            try:
-                kwargs['limit'] = int(tokens[i + 1])
-            except ValueError:
-                pass
-            i += 2
-        elif key == 'UNPUSHED_ONLY':
-            kwargs['unpushed_only'] = True
-            i += 1
-        elif key == 'WORKTREES_ONLY':
-            kwargs['worktrees_only'] = True
-            i += 1
-        else:
-            i += 1
-
-    return Query(**kwargs)
+    try:
+        while i < len(tokens):
+            key = tokens[i].upper()
+            if key in seen:
+                return None
+            seen.add(key)
+            if key in converters and i + 1 < len(tokens):
+                kwargs[key.lower()] = converters[key](tokens[i + 1])
+                i += 2
+            elif key in flags:
+                kwargs[key.lower()] = True
+                i += 1
+            else:
+                return None
+        return Query(**kwargs)
+    except (ValueError, TypeError, OverflowError):
+        return None
 
 
 def parse_natural_language(text):
@@ -142,6 +155,9 @@ def parse_natural_language(text):
     elif re.search(r'(scalon|zmergowan|merged)', low):
         state = 'merged'
 
+    if re.search(r'(advise|advice|porad|zaleceni|rekomendac)', low):
+        return Query('advise', raw_input=raw)
+
     # Domain 1: PRs & branches
     if re.search(r'(\bprs?\b|\bpr-[a-z0-9]+\b|pull\s*requests?|ga[łl][ęe]z|branch|\bmerg|scal|nieprzepchni)', low):
         return Query('prs', hours=hours if hours is not None else 24.0,
@@ -182,10 +198,8 @@ def parse(text):
     if not text or not text.strip():
         return None
     raw = text.strip()
-    if raw.upper().startswith('OBSERVE '):
-        dsl_query = parse_dsl(raw)
-        if dsl_query:
-            return dsl_query
+    if raw.upper().startswith('OBSERVE'):
+        return parse_dsl(raw)
     return parse_natural_language(raw)
 
 
@@ -204,6 +218,8 @@ def execute(query_or_text, root, depth=2, pr_limit=200, issue_limit=200, registr
     else:
         query = query_or_text
 
+    from .nl_contract import validate_query
+    validate_query(query)
     target = query.target
     data = None
     doc = ''
@@ -216,7 +232,7 @@ def execute(query_or_text, root, depth=2, pr_limit=200, issue_limit=200, registr
 
     elif target == 'audit':
         from . import audit
-        data = audit.scan(root, depth=depth, issue_limit=issue_limit,
+        data = audit.scan(root, depth=depth, issue_limit=query.issue_limit or issue_limit,
                           recent_hours=query.hours if query.hours != 24.0 else None,
                           worktrees_hours=query.hours,
                           worktrees_only=query.worktrees_only)
@@ -246,6 +262,17 @@ def execute(query_or_text, root, depth=2, pr_limit=200, issue_limit=200, registr
         data = catalog.scan(root, depth=depth)
         doc = catalog.markdown(data, limit=query.limit)
 
+    elif target == 'advise':
+        from . import advise
+        data = advise.advise(root, depth=depth, limit=query.limit,
+                             radar=query.radar, tier=query.tier)
+        if query.emit_planfile:
+            import json
+            data = advise.export_planfile_tickets(data, tier=query.tier)
+            doc = json.dumps(data, ensure_ascii=False, indent=2)
+        else:
+            doc = advise.markdown(data)
+
     return {
         'schema': SCHEMA,
         'status': 'ok',

src/monag/dsl_llm.py

@@ -24,7 +24,7 @@
 
 Grammar (exactly one line, conforming or nothing):
 OBSERVE <domain> [HOURS <n>] [STATE <open|merged|all>] [LIMIT <n>] [UNPUSHED_ONLY] [WORKTREES_ONLY]
-domain := prs | audit | status | resume | usage | catalog
+domain := prs | audit | status | resume | usage | catalog | advise
 
 Domains: prs = pull requests and branches, audit = planfile/GitHub coverage,
 status = agent processes and checkout snapshot, resume = worktree checkouts
@@ -94,6 +94,9 @@ def resolve(text, command=None, timeout=None):
     if rule_query is not None:
         return rule_query, provenance('rule', rule_query.to_dsl(), raw)
 
+    if raw.upper().startswith('OBSERVE'):
+        return None, provenance('none', None, raw)
+
     if command is None:
         command = os.environ.get('MONAG_LLM_COMMAND', '')
     if timeout is None:
@@ -110,10 +113,10 @@ def resolve(text, command=None, timeout=None):
     answer = run_provider(SYSTEM_PROMPT + '\nRequest: ' + raw, command, timeout)
     line = extract_observe_line(answer)
     if line is None or line.lower() == 'observe none':
-        return None, provenance('none', None, raw)
+        return None, dict(provenance('none', None, raw), attempted_engine='llm')
     query = dsl.parse_dsl(line)
     if query is None:
-        return None, provenance('none', None, raw)
+        return None, dict(provenance('none', None, raw), attempted_engine='llm')
     query.raw_input = raw
     return query, provenance('llm', query.to_dsl(), raw)