feat(prompts): native Muse tools (cron.create, subagent spawn) primary in envelope; harvester bridges native names to box ops

bin/prompt_envelope.py

@@ -46,6 +46,14 @@ def spawn_call(job_id, job_name, profile):
     return _tool("swarm.spawn", {"count": count, "task": task[:900], "label": (job_name or "job")[:60]})
 
 
+def native_followup_call(job_id, job_name, profile, agent):
+    """Muse-native one-shot cron (cron.create runonce) - bridged to followup.create."""
+    _, mins, _ = PROFILES[profile]
+    return _tool("cron.create", {
+        "kind": "runonce", "agent": agent, "in_m": mins,
+        "prompt": "Continue job %s (%s): verify with cron.runs and swarm.list then act or close." % (job_id, job_name)})
+
+
 def followup_call(job_id, job_name, profile, agent):
     _, mins, _ = PROFILES[profile]
     return _tool("followup.create", {
@@ -62,7 +70,7 @@ def wrap(job_name, job_id, agent, target, rendered):
     """Return the full DM body (without the [JOB id] prefix)."""
     profile = pick_profile(job_name)
     spawn = spawn_call(job_id, job_name, profile)
-    follow = followup_call(job_id, job_name, profile, agent)
+    follow = native_followup_call(job_id, job_name, profile, agent)
     top = (
         "WORK ORDER - ACTION REQUIRED, NOT INFORMATIONAL.\n"
         "Your FIRST output must be tool calls, not prose. Start now:\n"
@@ -78,11 +86,14 @@ def wrap(job_name, job_id, agent, target, rendered):
         "If you have not yet spawned subagents or set a follow-up timer, do it now:\n"
         "  %s\n"
         "  %s\n"
-        "Other tools: [TOOL health.check {}] [TOOL swarm.list {}] [TOOL swarm.status {\"swarm_id\": \"<id>\"}] "
+        "Native tools (preferred): cron.create (runonce/interval), cron.status, cron.view, cron.runs "
+        "(truth about what ran), cron.run, subagent spawn; hooks.dry_run before hooks.enable. "
+        "Native cron form: %s\n"
+        "Box fallback tools: [TOOL health.check {}] [TOOL swarm.list {}] [TOOL swarm.status {\"swarm_id\": \"<id>\"}] "
         "[TOOL swarm.results {\"swarm_id\": \"<id>\"}] [TOOL quality.check {}]\n"
         "Direct: curl -sk -X POST %s -H \"Authorization: Bearer <token>\" "
         "-H \"Content-Type: application/json\" -d '{\"op\":\"<op>\",\"args\":{...}}'\n"
-        "Thread: %s\n" % (spawn, follow, BOX_EXEC, thread_url(target))
+        "Thread: %s\n" % (spawn, follow, native_followup_call(job_id, job_name, profile, agent), BOX_EXEC, thread_url(target))
     )
     if not has_result:
         bottom += "Close with: [RESULT %s] <OK|FAILED>: <summary of tool outputs>\n" % job_id

bin/response-harvester.py

@@ -125,6 +125,48 @@ def detect_explicit_refusal(text):
     return any(re.search(pat, t) for pat in REFUSAL_PATTERNS)
 
 
+# Muse-native tool names (docs.muse-dev.online/capabilities.html) -> box exec ops.
+NATIVE_ALIASES = {
+    "cron.create": "followup.create",
+    "cron.runonce": "followup.create",
+    "cron.schedule": "followup.create",
+    "subagent.spawn": "swarm.spawn",
+    "subagents.spawn": "swarm.spawn",
+    "subagent.list": "swarm.list",
+    "subagent.status": "swarm.status",
+}
+
+
+def normalize_native_call(op, args):
+    """Bridge native Muse tool names/args to the box ops the exec daemon allows."""
+    op = NATIVE_ALIASES.get(op, op)
+    if not isinstance(args, dict):
+        return op, args
+    args = dict(args)
+    if op == "followup.create":
+        for k in ("kind", "mode", "schedule"):  # native cron fields the box op does not take
+            args.pop(k, None)
+        if "in_m" not in args:
+            for k in ("minutes", "in_minutes", "delay_m", "runonce_m"):
+                if k in args:
+                    args["in_m"] = args.pop(k)
+                    break
+        if "prompt" not in args:
+            for k in ("body", "task", "message", "text"):
+                if k in args:
+                    args["prompt"] = args.pop(k)
+                    break
+    elif op == "swarm.spawn":
+        if "task" not in args:
+            for k in ("prompt", "body", "instructions"):
+                if k in args:
+                    args["task"] = args.pop(k)
+                    break
+        if "count" not in args and "n" in args:
+            args["count"] = args.pop("n")
+    return op, args
+
+
 def parse_tool_calls(text):
     """
     Extract structured tool/exec calls from assistant messages.
@@ -166,7 +208,7 @@ def parse_tool_calls(text):
     # De-duplicate identical calls within one message (envelope repeats the spawn
     # call at top and bottom; an agent echoing both must not run it twice).
     seen, unique = set(), []
-    for op, args in calls:
+    for op, args in [normalize_native_call(o, a) for o, a in calls]:
         key = (op, json.dumps(args, sort_keys=True, default=str))
         if key in seen:
             continue