Repair production automation truth and schedule cleanup

src/activity_core/activities.py

@@ -352,7 +352,20 @@ async def evaluate_instructions(payload: dict) -> dict:
     reports: list[dict] = []
     for raw_instruction in instructions:
         try:
-            instruction = InstructionDef.model_validate(raw_instruction)
+            instruction_data = dict(raw_instruction)
+            output_schema = instruction_data.get("output_schema")
+            if isinstance(output_schema, str) and output_schema.startswith(
+                ("custodian://", "activity-core://")
+            ):
+                # Resolve deployment-neutral URIs at the workflow/activity
+                # boundary.  The pure rules package only accepts filesystem
+                # paths and must not import runtime integration modules.
+                from activity_core.runtime_paths import resolve_runtime_path
+
+                instruction_data["output_schema"] = str(
+                    resolve_runtime_path(output_schema)
+                )
+            instruction = InstructionDef.model_validate(instruction_data)
         except Exception as exc:
             activity.logger.warning("instruction definition invalid — %s", exc)
             continue

src/activity_core/models.py

@@ -182,6 +182,13 @@ class ActivityDefinition(BaseModel):
     instructions: list[InstructionDef] = Field(default_factory=list)
     # Legacy — ignored when rules is non-empty
     task_templates: list[TaskTemplate] = Field(default_factory=list)
-    dedupe_key_strategy: Literal["skip", "catchup", "compress"] = Field(default="skip")
+    dedupe_key_strategy: Literal["skip", "catchup", "compress"] = Field(
+        default="skip",
+        description=(
+            "Legacy persistence compatibility metadata; runtime scheduling uses "
+            "trigger_config.misfire_policy and this field does not deduplicate content"
+        ),
+        deprecated=True,
+    )
     version: int = Field(default=1, ge=1)
     status: str = Field(default="active")

src/activity_core/rules/executor.py

@@ -877,9 +877,7 @@ def _load_output_schema(schema_path: str) -> dict[str, Any] | None:
     if not schema_path:
         return None
 
-    from activity_core.runtime_paths import resolve_runtime_path
-
-    path = resolve_runtime_path(schema_path)
+    path = Path(schema_path).expanduser()
     if not path.exists():
         return None
 

src/activity_core/sync_activity_definitions.py

@@ -72,6 +72,9 @@ async def sync(session_factory: async_sessionmaker[AsyncSession]) -> int:
                         task_templates=[],
                         rules_json=d.rules,
                         instructions_json=d.instructions,
+                        # Legacy DB compatibility only. Schedule behavior comes
+                        # from trigger_config.misfire_policy; this value does not
+                        # suppress repeated content across nominal fire times.
                         dedupe_key_strategy="skip",
                         version=1,
                     )

src/activity_core/sync_schedules.py

@@ -24,7 +24,12 @@ from temporalio.client import Client
 
 from activity_core.models import ActivityDefinition, CronTriggerConfig, ScheduledTriggerConfig
 from activity_core.orm import ActivityDefinition as ActivityDefinitionRow
-from activity_core.schedule_manager import delete_schedule, list_schedules, upsert_schedule
+from activity_core.schedule_manager import (
+    cancel_scheduled,
+    delete_schedule,
+    list_schedules,
+    upsert_schedule,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -107,7 +112,14 @@ async def sync_schedule_rows(
         valid_schedule_activity_ids.add(_valid_schedule_activity_id(defn))
 
         try:
-            await upsert_schedule(client, defn)
+            if isinstance(defn.trigger_config, ScheduledTriggerConfig) and not defn.enabled:
+                # A disabled one-shot has no future recurrence to preserve. A
+                # completed Temporal schedule can reject in-place updates with
+                # ScheduleAlreadyRunningError, leaving every later sync noisy.
+                # Delete it idempotently instead; re-enabling recreates it.
+                await cancel_scheduled(client, defn.id)
+            else:
+                await upsert_schedule(client, defn)
         except Exception as exc:  # noqa: BLE001 — continue reconcile for other rows
             result.errors += 1
             detail = f"{defn.id} ({defn.name}): {type(exc).__name__}: {exc}"
@@ -120,7 +132,10 @@ async def sync_schedule_rows(
             logger.info("upserted schedule for activity %s (%s)", defn.id, defn.name)
         else:
             result.paused += 1
-            logger.info("upserted paused schedule for disabled activity %s", defn.id)
+            if isinstance(defn.trigger_config, ScheduledTriggerConfig):
+                logger.info("removed schedule for disabled one-shot activity %s", defn.id)
+            else:
+                logger.info("upserted paused schedule for disabled activity %s", defn.id)
 
     # Tombstone cleanup: remove Temporal Schedules with no matching DB row.
     try: