Merge remote-tracking branch 'origin/main' into feat/247-foreign-workflows-pr2

PR 3's repair skips a file whose recorded saver's branch reads a
switched-off PromptManager node. On the monitor the same shape never reaches
the foreign path: a literal switch is the opted-out path's and a wired one
the node's own run's. Nothing is stored at settle or after the run, even
from a recorded run holding the off text.

TESTING.md

@@ -3238,7 +3238,8 @@ until this browser dismissed the 3.2.0 update notice (`pm_update_notice_dismisse
 the legacy admin page's key); Later dismisses it without scanning, and Review scan
 dismisses it as the legacy Scan does. "+ Add source" asks validate-source, then adds
 the folder to the host draft of `gallery_root_paths`; Remove takes one out; "Watch for
-new outputs" (`monitoring_enabled`) and Worker threads (`worker_threads`) are host
+new outputs" (`monitoring_enabled`), "Save prompts from other workflows"
+(`save_foreign_runs`, D158) and Worker threads (`worker_threads`) are host
 drafts too, and Save writes them with the revision. A save the server refuses because a
 scan runs toasts "Another job is running: <title>." with a link to that job (D57) and
 keeps the drafts; a save refused for one source says the server's message as the status
@@ -3487,7 +3488,7 @@ the connection's `total_changes` is 0 at close and the data reads back unchanged
 `-shm` bytes are not compared: every WAL reader rewrites its read marks there, so no
 reader can keep it byte-identical (`test_health_check_never_writes`).
 
-Recorded deviations (plan D60–D75, D79–D82 renumbered D92–D107, D109–D110 and D111–D114; D121–D134 assigned). D94,
+Recorded deviations (plan D60–D75, D79–D82 renumbered D92–D107, D109–D110 and D111–D114; D121–D134 and D158 assigned). D94,
 the per-section hand-off panel, was retired by S12 (#329): every section is native.
 
 - **D92** (shell): no "Sample configuration" chip. The Save status is truthful: "No
@@ -3976,6 +3977,26 @@ the per-section hand-off panel, was retired by S12 (#329): every section is nati
   Cancel-only review matches the concept review shell` (system-tagging-concept.spec) and
   the batch tests of workspace-system-tagging.spec; the mutation check renames each grid
   label in turn.
+- **D158** (Sources, #247 D-Q4): the concept has no switch for prompts from workflows
+  without a PromptManager node; the maintainer chose a visible switch over a config key.
+  A `.setting-row` in the concept's toggle markup follows "Watch for new outputs":
+  "Save prompts from other workflows" (`#foreign-runs`, host key `save_foreign_runs`,
+  config.json `gallery.monitoring.save_foreign_runs`, default on), "New outputs of
+  workflows without a PromptManager node add their prompt to the library when it is
+  certain.". It is saved with the revision like the other host rows and applied at once
+  (the monitor reads it for every file), so a save names no restart. While "Watch for
+  new outputs" is drafted off, a `p.small` under the description says "Applies while
+  Watch for new outputs is on." (the switch's `aria-describedby`; the value is kept) and
+  the switch is disabled, except while a source still reads "watching until restart"
+  (D128): the running observer still delivers files then, so the value can be set. The
+  server stores no foreign prompt while `monitoring_enabled` is saved off, whatever the
+  observer does.
+  Pinned by `sources matches the concept` (the mapping inserts the row after the
+  monitoring row), `sources comparison detects mutations` (a missing row and a changed
+  label are caught), `D158: Save prompts from other workflows saves save_foreign_runs
+  live; off while monitoring is off (#247)` (keyboard on desktop, a tap on touch) and
+  `D158: saved off while the monitor still watches until a restart, the foreign row can
+  still be set (D128)` (workspace-system-sources.spec) and the system-sources node test.
 - **D109** (Logs, S12, #329): the concept's Logs surface on the live server. The toolbar
   keeps "Level", Refresh and the download (renamed "Download log": the active log,
   `/logs/download/prompt_manager.log`, through the api client), with all six minimum
@@ -5903,6 +5924,70 @@ Tests: `tests/test_file_prompt.py` (one test per ladder step and junk class),
 saver, a legacy-linked file left as it is, `_BatchHistory.recorded`) and
 `tests/test_recorded_saver_anchor.py` `TestInspectionResolution`.
 
+**In the monitor.** An output whose workflow has no positive PromptManager node
+(`_find_prompt_from_metadata` says `foreign`) used to be logged and skipped. It now
+goes through the same ladder with the `MONITOR` policy (`utils/foreign_runs.py`, the
+handler's `_ingest_foreign`, `_recheck_foreign`, `_store_foreign`):
+
+- **At settle.** The ladder runs without history. Certain (`pair` or `positive`): the
+  prompt is stored and the file linked at once. `history` (an undecided graph inside
+  ComfyUI's output folder): the file is registered in its own waiting set, then the
+  ladder runs again at once with the file's recorded output, because the run may have
+  been sealed before the file settled. When that lookup finds a record, its answer is
+  final: certain is stored, anything else is logged "No certain prompt for <file>;
+  left unlinked" and the file stops waiting (no seal is coming). When it finds none,
+  only the inspector's or the recorded saver's answer is stored; anything else (the
+  legacy floor included) keeps waiting for the re-check. Anything else at the first
+  pass is logged "No certain prompt for <file>; left unlinked".
+- **After the run ends.** When the run's outputs are sealed, each waiting foreign file
+  gets one re-check (`ForeignRecheck(reference)`) with the sealed output reference,
+  so history is asked with the spelling it recorded (a Windows folder whose case
+  differs from the disk still links). Not certain: "No certain prompt for <file>
+  after its run ended; left unlinked". An ambiguous inspector answer is never given
+  the legacy floor in the monitor (D-Q3); an untraceable one still is.
+- **Waiting sets.** `_awaiting_foreign` and `_foreign_due` sit beside the keyless sets
+  with the same bounds (1000 files, one hour) but are separate, so a large foreign run
+  never evicts a waiting keyless file. An expired entry is logged "No recorded run
+  for <file>; left unlinked", an evicted one (waiting or due) "Too many files wait for
+  their run; <file> left unlinked", both after the lock is released. A due re-check
+  is dropped only when it was scheduled, never a newer reference for the same file. `awaiting_foreign()` lists them;
+  `shutdown()` and `close()` clear them, and a closed handler re-checks nothing.
+- **Store.** One write through `survive_switch` (rerun after a Move library): a pair
+  through `ingest_resolved_pair(..., marks=RUN_MARKS)`, a positive through
+  `store_scanned_prompt_or_raise(..., RUN_MARKS)`. A created prompt carries category
+  `scanned`, tag `auto-scanned` and the note `Saved from a workflow run: <file>`; a
+  file linked to another prompt is `conflict` and keeps its link. Logged at info
+  "Foreign run: <status> prompt <id> for <file> (<source>)"; a failure is logged and
+  the file stays unlinked.
+- **Never.** No tracker FIFO entry is read or consumed, nothing is counted (run count
+  stays 0, no `last_used_at`), and nothing is queued for Needs review.
+- **The setting.** `GalleryConfig.SAVE_FOREIGN_RUNS` (`gallery.monitoring.
+  save_foreign_runs`, default true, only a bool applies) is read for every file, at
+  settle, at the re-check and again just before the store, together with
+  `MONITORING_ENABLED`: either saved off stops foreign writes at once (the running
+  observer watches until a restart, D128); off logs "No PromptManager prompt in
+  workflow, skipping image" as before. System › Sources & scans "Save prompts from other workflows" (D158)
+  saves it through `POST /prompt_manager/settings` with no restart field; the
+  diagnostics report lists it.
+
+Tests: `tests/test_run_attribution.py` `TestMonitorForeignRuns` (dataset branches
+linked after the seal, a seal before settle, a name two runs reported, the setting
+off at settle and at the re-check, a plain workflow linked at settle with no run
+count and the tracker queue untouched, the sealed reference's spelling, the separate
+bound, expiry, a seal while the pending set is full, shutdown, a failed store, no
+record at settle keeps waiting, a record without an answer ends the wait, monitoring
+saved off, the setting read before the store, a seal between registering and the
+settle lookup, a newer due reference kept, evictions logged, one link per file),
+`tests/test_image_linking.py` `TestForeignWorkflows`, `tests/test_image_monitor_singleton.py`
+(close drops a foreign re-check), `tests/test_config.py`,
+`tests/test_settings_sources.py` `TestSaveForeignRunsSetting`, and the real-host
+`tests/comfyui/image-attribution.spec.mjs` case "an output of a workflow without a
+PromptManager node creates and links its prompt; a second run reuses it" (ComfyUI's
+own `CLIPTextEncode`, `KSampler` and `SaveImage`), and "each branch of a
+shared-prefix workflow without a PromptManager node links to its own prompt after the
+run" (two branches, one save prefix: the recorded saver names each file's branch).
+The fixture route's GET also lists `foreign`, the waiting foreign files.
+
 ## Linked prompt text (#343)
 
 A link drives `text` on every PromptManager node, and `negative_text` on the Pair,

py/api/diagnostics_report.py

@@ -213,6 +213,8 @@ def public_config():
             "max_image_age_days": GalleryConfig.MAX_IMAGE_AGE_DAYS,
             "max_concurrent_processing": GalleryConfig.MAX_CONCURRENT_PROCESSING,
             "metadata_extraction_timeout": GalleryConfig.METADATA_EXTRACTION_TIMEOUT,
+            # config.json gallery.monitoring.save_foreign_runs (#247)
+            "save_foreign_runs": GalleryConfig.SAVE_FOREIGN_RUNS,
         },
         "integrations": {
             "lora_manager_enabled": bool(IntegrationConfig.LORA_MANAGER_ENABLED),

py/api/settings_routes.py

@@ -8,7 +8,8 @@ is shared with the LoRA integration's save.
 GET also reports the image sources and the monitor (``sources.py``). A save
 that changes the sources runs as a ``source_change`` registry hold, so it is
 refused while a record-changing job (a scan) runs. ``monitoring_enabled``
-takes effect at the next restart. ``POST /prompt_manager/settings/validate-source``
+takes effect at the next restart; ``save_foreign_runs`` (#247) at once, since
+the monitor reads it for every file. ``POST /prompt_manager/settings/validate-source``
 checks one path and writes nothing.
 """
 
@@ -43,6 +44,7 @@ _SETTINGS_FILE_KEYS = {
     "INFINITE_SCROLL": ("web_ui", "infinite_scroll"),
     "WORKER_THREADS": ("performance", "worker_threads"),
     "MONITORING_ENABLED": ("gallery", "monitoring", "enabled"),
+    "SAVE_FOREIGN_RUNS": ("gallery", "monitoring", "save_foreign_runs"),
     "SAFETY_BACKUP": ("maintenance", "safety_backup"),
     "THUMBNAIL_FORMAT": ("thumbnails", "format"),
     "THUMBNAIL_QUALITY": ("thumbnails", "quality"),
@@ -58,7 +60,7 @@ _SETTINGS_FILE_KEYS = {
 # The thumbnail settings a running Convert outdated job fixes (#317).
 _THUMBNAIL_SETTINGS = ("THUMBNAIL_FORMAT", "THUMBNAIL_QUALITY", "THUMBNAIL_EDGE")
 # Attributes that live on GalleryConfig; the others on PromptManagerConfig.
-_GALLERY_SETTINGS = frozenset({"MONITORING_ENABLED"})
+_GALLERY_SETTINGS = frozenset({"MONITORING_ENABLED", "SAVE_FOREIGN_RUNS"})
 # Attributes that live on AutotagConfig, named without the AUTOTAG_ prefix.
 _AUTOTAG_SETTINGS = frozenset(
     name for name in _SETTINGS_FILE_KEYS if name.startswith("AUTOTAG_")
@@ -264,6 +266,7 @@ class SettingsRoutesMixin:
                     "thumbnail_encoders": encoders,
                     "settings": {
                         "monitoring_enabled": GalleryConfig.MONITORING_ENABLED,
+                        "save_foreign_runs": GalleryConfig.SAVE_FOREIGN_RUNS,
                         "result_timeout": PromptManagerConfig.RESULT_TIMEOUT,
                         "webui_display_mode": PromptManagerConfig.WEBUI_DISPLAY_MODE,
                         "infinite_scroll": PromptManagerConfig.INFINITE_SCROLL,
@@ -626,6 +629,10 @@ class SettingsRoutesMixin:
             if not isinstance(data["monitoring_enabled"], bool):
                 raise ValueError("monitoring_enabled must be true or false")
             updates["MONITORING_ENABLED"] = data["monitoring_enabled"]
+        if "save_foreign_runs" in data:
+            if not isinstance(data["save_foreign_runs"], bool):
+                raise ValueError("save_foreign_runs must be true or false")
+            updates["SAVE_FOREIGN_RUNS"] = data["save_foreign_runs"]
         if "safety_backup" in data:
             if not isinstance(data["safety_backup"], bool):
                 raise ValueError("safety_backup must be true or false")

py/config.py

@@ -194,6 +194,8 @@ class GalleryConfig:
 
     Attributes:
         MONITORING_ENABLED (bool): Enable/disable automatic image monitoring
+        SAVE_FOREIGN_RUNS (bool): The monitor stores the certain prompt of an
+            output whose workflow has no PromptManager node (#247)
         MONITORING_DIRECTORIES (List[str]): Directories to monitor for new images
         SUPPORTED_EXTENSIONS (List[str]): Image file extensions to process
         PROCESSING_DELAY (float): Delay in seconds before processing new files
@@ -210,6 +212,8 @@ class GalleryConfig:
 
     # Image monitoring settings
     MONITORING_ENABLED = True
+    # Read by the monitor for every file, so a save applies without a restart
+    SAVE_FOREIGN_RUNS = True
     MONITORING_DIRECTORIES = []  # Auto-detect if empty
     SUPPORTED_EXTENSIONS = [".png", ".jpg", ".jpeg", ".webp", ".gif"]
     PROCESSING_DELAY = 2.0  # Seconds to wait before processing new files
@@ -251,6 +255,7 @@ class GalleryConfig:
         return {
             "monitoring": {
                 "enabled": cls.MONITORING_ENABLED,
+                "save_foreign_runs": cls.SAVE_FOREIGN_RUNS,
                 "directories": cls.MONITORING_DIRECTORIES,
                 "extensions": cls.SUPPORTED_EXTENSIONS,
                 "processing_delay": cls.PROCESSING_DELAY,
@@ -303,6 +308,14 @@ class GalleryConfig:
                 config_logger.warning(
                     "Ignoring gallery monitoring enabled: expected true or false"
                 )
+        if "save_foreign_runs" in monitoring:
+            if isinstance(monitoring["save_foreign_runs"], bool):
+                cls.SAVE_FOREIGN_RUNS = monitoring["save_foreign_runs"]
+            else:
+                config_logger.warning(
+                    "Ignoring gallery monitoring save_foreign_runs: expected true "
+                    "or false"
+                )
         if "directories" in monitoring:
             directories = cls._validated_directories(monitoring["directories"])
             if directories is not None:

tests/browser/system-concept.spec.mjs

@@ -617,10 +617,21 @@ const WATCHED = new Set(['watching', 'stopping']);
 const REFRESH_FILES = { title: 'Refresh files', description: 'Read every source again to find files added or removed outside ComfyUI.',
     label: 'Refresh' };
 
+// D158 (#247): the row production adds directly below "Watch for new outputs", in that row's markup.
+const FOREIGN_RUNS = { id: 'foreign-runs', label: 'Save prompts from other workflows',
+    description: 'New outputs of workflows without a PromptManager node add their prompt to the library when it is certain.' };
+
+/** D158: the concept's "Watch for new outputs" row as the inserted "Save prompts from other workflows" row. */
+function foreignRunsRow(monitoringRow, settings) {
+    return { ...monitoringRow, text: [['label', FOREIGN_RUNS.label, FOREIGN_RUNS.id], ['p', FOREIGN_RUNS.description, null]],
+        control: { ...monitoringRow.control, id: FOREIGN_RUNS.id, checked: settings.save_foreign_runs } };
+}
+
 /** The concept's Sources with the recorded deviations applied, given the server's real settings and sources. */
 function mapSourcesConcept(reference, { settings, sources }) {
     const [sample] = reference.sources;
     const scan = sample.actions[0];
+    const rowAt = reference.surface.tree.indexOf('div.setting-row');
     return {
         ...reference,
         // The real sources replace the two samples (without the "sample path" note). D98: each configured one has Remove,
@@ -628,13 +639,15 @@ function mapSourcesConcept(reference, { settings, sources }) {
         sources: sources.effective.map(source => ({ ...sample, title: source.label,
             description: `${source.path} · ${WATCH_TEXT[source.watch]}`,
             actions: [{ ...scan, text: WATCHED.has(source.watch) ? 'Scan changes' : 'Scan folder' }, { ...scan, text: 'Remove' }] })),
-        rows: reference.rows.map(row => {
-            if (row.control.id === 'monitoring') return { ...row, control: { ...row.control, checked: settings.monitoring_enabled } };
+        rows: reference.rows.flatMap(row => {
+            if (row.control.id === 'monitoring') return [{ ...row, control: { ...row.control, checked: settings.monitoring_enabled } },
+                foreignRunsRow(row, settings)];
             // D113: the range ends at the server's core count (the concept's fixed 16) and shows the real worker count.
-            return { ...row, control: { ...row.control, max: String(settings.cpu_count), value: String(settings.worker_threads) } };
+            return [{ ...row, control: { ...row.control, max: String(settings.cpu_count), value: String(settings.worker_threads) } }];
         }),
-        // D98: "Refresh files" closes the surface.
-        surface: { ...reference.surface, tree: [...reference.surface.tree, 'div.operation'] },
+        // D158: the inserted row after the first setting row; D98: "Refresh files" closes the surface.
+        surface: { ...reference.surface, tree: [...reference.surface.tree.slice(0, rowAt + 1), 'div.setting-row',
+            ...reference.surface.tree.slice(rowAt + 1), 'div.operation'] },
         operations: [...reference.operations, { ...reference.operations[0], title: REFRESH_FILES.title,
             description: REFRESH_FILES.description, actions: [{ ...reference.operations[0].actions[0], text: REFRESH_FILES.label }] }],
     };
@@ -688,10 +701,12 @@ test('sources comparison detects mutations', async ({ page, request }) => {
     const expected = mapSourcesConcept(concept, server);
     const production = await productionSources(page, 'dark', { notice: true });
     expect(diffInventory(expected, production)).toEqual([]);
-    // Unmapped, the concept differs only in the real sources, the D98 operation and the D113 range.
+    // Unmapped, the concept differs only in the real sources, the D98 operation, the D158 row (which moves the D113
+    // range to the third row) and the D113 range.
     expect(diffInventory(concept, production).map(item => item.path.split('.').slice(0, 2).join('.'))
         .filter((path, index, all) => all.indexOf(path) === index).sort())
-        .toEqual(['operations.2', 'operations.length', 'rows.1', 'sources.0', 'sources.1', 'surface.tree'].sort());
+        .toEqual(['operations.2', 'operations.length', 'rows.1', 'rows.2', 'rows.length', 'sources.0', 'sources.1',
+            'surface.tree'].sort());
     const measure = () => page.evaluate(measureSources, '#workspace-main');
     const paths = async () => diffInventory(expected, await measure()).map(item => item.path);
     // A removed notice and a removed source row.
@@ -705,6 +720,13 @@ test('sources comparison detects mutations', async ({ page, request }) => {
     await page.evaluate(() => { document.querySelector('#workspace-main label[for="monitoring"]').textContent = 'Watch for outputs'; });
     expect(await paths()).toEqual(['rows.0.text.0.1']);
     await page.evaluate(() => { document.querySelector('#workspace-main label[for="monitoring"]').textContent = 'Watch for new outputs'; });
+    // D158: the inserted row is compared too: a missing one and a changed label are caught.
+    const undoForeign = await mutate(page, { remove: ['#workspace-main .setting-row:has(#foreign-runs)'] });
+    expect(await paths()).toEqual(expect.arrayContaining(['rows.length', 'surface.tree.length']));
+    await undoForeign();
+    await page.evaluate(() => { document.querySelector('#workspace-main label[for="foreign-runs"]').textContent = 'Save other prompts'; });
+    expect(await paths()).toEqual(['rows.1.text.0.1']);
+    await page.evaluate(() => { document.querySelector('#workspace-main label[for="foreign-runs"]').textContent = 'Save prompts from other workflows'; });
     await page.evaluate(() => { document.querySelector('#workspace-main .panel-heading button').textContent = 'Add source'; });
     expect(await paths()).toEqual(['heading.action.text']);
     await page.evaluate(() => { document.querySelector('#workspace-main .panel-heading button').textContent = '+ Add source'; });

tests/browser/workspace-system-sources.spec.mjs

@@ -268,6 +268,71 @@ test('Watch for new outputs saves monitoring_enabled; a restart applies it', asy
     await expect(sourceList(page).locator('.operation p')).toHaveText(['output · watching', 'input · watching']);
 });
 
+test('D158: Save prompts from other workflows saves save_foreign_runs live; off while monitoring is off (#247)', async ({ page, request }, testInfo) => {
+    try {
+        const layout = await openSources(page);
+        const rows = layout.locator('section.surface > .setting-row');
+        await expect(rows.locator(':scope > div > label')).toHaveText(['Watch for new outputs', 'Save prompts from other workflows', 'Worker threads']);
+        const foreign = layout.getByRole('switch', { name: 'Save prompts from other workflows', exact: true });
+        const foreignRow = rows.filter({ has: page.locator('#foreign-runs') });
+        await expect(foreignRow.locator(':scope > div > p')).toHaveText('New outputs of workflows without a PromptManager node add their prompt to the library when it is certain.');
+        await expect(foreign).toBeChecked();
+        await expect(foreign).toBeEnabled();
+        // Keyboard on desktop, a tap on the touch project.
+        if (testInfo.project.name === 'touch') await foreign.tap();
+        else {
+            await foreign.focus();
+            await page.keyboard.press('Space');
+        }
+        await expect(foreign).not.toBeChecked();
+        await expect(statusOf(page)).toHaveText('Unsaved changes.');
+        const saved = page.waitForResponse(response => response.request().method() === 'POST' && response.url().endsWith(SETTINGS_URL));
+        await saveButton(page).click();
+        const response = await saved;
+        expect(response.request().postDataJSON()).toEqual(expect.objectContaining({ save_foreign_runs: false }));
+        expect((await response.json()).restart_required_fields).toEqual([]);
+        // Applied live: the monitor reads it for every file, so no restart is named.
+        await expect(statusOf(page)).toHaveText('Saved to the server configuration.');
+        expect((await serverJson(request, SETTINGS_URL)).settings.save_foreign_runs).toBe(false);
+        // With Watch for new outputs drafted off nothing is watched: the row is disabled and its note says why.
+        const monitoring = layout.getByRole('switch', { name: 'Watch for new outputs', exact: true });
+        await monitoring.uncheck();
+        await expect(foreign).toBeDisabled();
+        const note = foreignRow.locator('p.small');
+        await expect(note).toHaveText('Applies while Watch for new outputs is on.');
+        await expect(foreign).toHaveAttribute('aria-describedby', await note.getAttribute('id'));
+        await monitoring.check();
+        await expect(foreign).toBeEnabled();
+        await expect(note).toHaveCount(0);
+        await expect(foreign).not.toHaveAttribute('aria-describedby', /./);
+        await expect(statusOf(page)).toHaveText('No unsaved changes.');
+    } finally {
+        // The server is shared by the spec's tests: put the default back whatever happened.
+        await request.post(SETTINGS_URL, { data: { save_foreign_runs: true } });
+    }
+});
+
+test('D158: saved off while the monitor still watches until a restart, the foreign row can still be set (D128)', async ({ page, request }) => {
+    await expect(await request.post('/__test__/system/monitor', { data: { watching: ['output', 'input'] } })).toBeOK();
+    try {
+        const layout = await openSources(page);
+        await layout.getByRole('switch', { name: 'Watch for new outputs', exact: true }).uncheck();
+        await saveButton(page).click();
+        await expect(sourceList(page).locator('.operation p').first()).toHaveText('output · watching until restart');
+        const foreign = layout.getByRole('switch', { name: 'Save prompts from other workflows', exact: true });
+        await expect(foreign).toBeEnabled();
+        await expect(layout.locator('.setting-row').filter({ has: page.locator('#foreign-runs') }).locator('p.small'))
+            .toHaveText('Applies while Watch for new outputs is on.');
+        await foreign.uncheck();
+        const saved = page.waitForRequest(item => item.method() === 'POST' && item.url().endsWith(SETTINGS_URL));
+        await saveButton(page).click();
+        expect((await saved).postDataJSON()).toEqual({ save_foreign_runs: false, expected_revision: expect.any(String) });
+        await expect(statusOf(page)).toHaveText('Saved to the server configuration.');
+    } finally {
+        await request.post(SETTINGS_URL, { data: { save_foreign_runs: true } });
+    }
+});
+
 test('worker range max is cpu_count', async ({ page, request }) => {
     const { settings } = await serverJson(request, SETTINGS_URL);
     const layout = await openSources(page);

tests/comfyui/fixtures/ledger_nodes.py

@@ -334,8 +334,15 @@ async def image_monitor(request: web.Request) -> web.Response:
 
 @PromptServer.instance.routes.get("/pm_execution_fixture/monitor")
 async def image_monitor_state(request: web.Request) -> web.Response:
-    """File names of keyless images waiting for their run's recorded outputs."""
+    """File names of images waiting for their run's recorded outputs: keyless
+    ones (``awaiting``) and foreign ones (``foreign``, #247)."""
     monitor = _plugin_module("utils.image_monitor")._monitor_instance
     handler = getattr(monitor, "handler", None)
     waiting = handler.awaiting_history() if handler is not None else []
-    return web.json_response({"awaiting": [os.path.basename(p) for p in waiting]})
+    foreign = handler.awaiting_foreign() if handler is not None else []
+    return web.json_response(
+        {
+            "awaiting": [os.path.basename(p) for p in waiting],
+            "foreign": [os.path.basename(p) for p in foreign],
+        }
+    )

tests/comfyui/image-attribution.spec.mjs

@@ -1,6 +1,7 @@
 // #342 on the real pinned host: the image monitor links an output only to a prompt
 // that provably produced it. An unknown node (PMHistoryCached) stands in for the
-// maintainer's LLM; KSampler is the fixture's deterministic stand-in.
+// maintainer's LLM; KSampler is the fixture's deterministic stand-in. #247: an output
+// of a workflow without a PromptManager node gets its own prompt.
 import { randomUUID } from 'node:crypto';
 import { execFile } from 'node:child_process';
 import { promisify } from 'node:util';
@@ -141,6 +142,68 @@ test('an output saved without metadata links once its recorded run names a store
     }
 });
 
+// #247: no PromptManager node at all, only ComfyUI's own text encoders, sampler and
+// saver (the fixture's inputs and decoder stand in for the model files).
+function foreignGraph({ text, seed = 1, prefix }) {
+    return {
+        '10': { class_type: 'PMHistoryInputs', inputs: {} },
+        '20': { class_type: 'CLIPTextEncode', inputs: { text, clip: ['10', 0] } },
+        '30': { class_type: 'CLIPTextEncode', inputs: { text: '', clip: ['10', 0] } },
+        '40': { class_type: 'KSampler', inputs: { model: ['10', 1], latent_image: ['10', 2], positive: ['20', 0],
+            negative: ['30', 0], seed, steps: 1, cfg: 1, sampler_name: 'euler', scheduler: 'normal', denoise: 1 } },
+        '60': { class_type: 'PMHistoryImages', inputs: { value: 64, batch: 1, latent: ['40', 0] } },
+        '70': { class_type: 'SaveImage', inputs: { images: ['60', 0], filename_prefix: prefix } },
+    };
+}
+
+test('an output of a workflow without a PromptManager node creates and links its prompt; a second run reuses it', async ({ execution }, testInfo) => {
+    const text = `foreign lighthouse at dusk ${testInfo.project.name} ${randomUUID()}`;
+    const first = await saved(execution.request, foreignGraph({ text, prefix: 'foreign/run' }));
+    await linkedTo(execution, first, text);
+    // Only the seed changes: a new file, the same text, so the same stored prompt.
+    const second = await saved(execution.request, foreignGraph({ text, seed: 2, prefix: 'foreign/run' }));
+    await linkedTo(execution, second, text);
+    const state = await readLinks(execution.databasePath, [first, second], [text]);
+    expect(state).toEqual({ links: { [first]: text, [second]: text }, stored: { [text]: true } });
+    // Decided from the graph at settle: nothing waits for the run's recorded outputs.
+    expect((await (await execution.request.get('/pm_execution_fixture/monitor')).json()).foreign).toEqual([]);
+});
+
+// CUJ 1: two branches saved under one prefix, so neither the graph nor the file name
+// names a file's branch; the run's recorded saver does, once its outputs are sealed.
+function sharedPrefixGraph(texts) {
+    const nodes = {
+        '10': { class_type: 'PMHistoryInputs', inputs: {} },
+        '30': { class_type: 'CLIPTextEncode', inputs: { text: '', clip: ['10', 0] } },
+    };
+    texts.forEach((text, index) => {
+        const base = (index + 1) * 100;
+        nodes[String(base)] = { class_type: 'CLIPTextEncode', inputs: { text, clip: ['10', 0] } };
+        nodes[String(base + 1)] = { class_type: 'KSampler', inputs: { model: ['10', 1], latent_image: ['10', 2],
+            positive: [String(base), 0], negative: ['30', 0], seed: index + 1, steps: 1, cfg: 1, sampler_name: 'euler',
+            scheduler: 'normal', denoise: 1 } };
+        nodes[String(base + 2)] = { class_type: 'PMHistoryImages', inputs: { value: 64, batch: 1, latent: [String(base + 1), 0] } };
+        nodes[String(base + 3)] = { class_type: 'SaveImage', inputs: { images: [String(base + 2), 0], filename_prefix: 'foreign/dataset' } };
+    });
+    return nodes;
+}
+
+test('each branch of a shared-prefix workflow without a PromptManager node links to its own prompt after the run', async ({ execution }, testInfo) => {
+    const tag = `${testInfo.project.name} ${randomUUID()}`;
+    const texts = [`dataset branch one harbor ${tag}`, `dataset branch two meadow ${tag}`];
+    const id = randomUUID();
+    const response = await execution.request.post('/prompt', { data: { prompt: sharedPrefixGraph(texts), prompt_id: id } });
+    expect(response.ok()).toBe(true);
+    let item;
+    await expect.poll(async () => {
+        item = (await (await execution.request.get(`/history/${id}`)).json())[id];
+        return item?.status?.messages?.some(([kind]) => kind === 'execution_success') || false;
+    }, { timeout: 15000 }).toBe(true);
+    const files = ['103', '203'].map(node => item.outputs[node].images[0].filename);
+    for (const [index, file] of files.entries()) await linkedTo(execution, file, texts[index]);
+    expect((await (await execution.request.get('/pm_execution_fixture/monitor')).json()).foreign).toEqual([]);
+});
+
 // #332 on the real host: the migration ran at startup and the image category routes
 // set, list and remove an own category on a generated output.
 test('an output gets its own image category and loses it again', async ({ execution }, testInfo) => {

tests/fixtures/legacy_contract_snapshot.json

@@ -192,6 +192,7 @@
     "settings.monitoring_enabled",
     "settings.result_timeout",
     "settings.safety_backup",
+    "settings.save_foreign_runs",
     "settings.thumbnail_edge",
     "settings.thumbnail_format",
     "settings.thumbnail_quality",

tests/js/workspace-system-sources.test.js

@@ -157,3 +157,20 @@ test('a valid configured source missing from the effective list is another spell
     assert.deepEqual(sourceRows({ sources, saved, current: saved }).rows.at(-1), { key: 'output/.', title: 'output',
         description: 'output/. · same folder as another source', scan: null, remove: 'output/.' });
 });
+
+test('D158: "Save prompts from other workflows" drafts save_foreign_runs, disabled while monitoring is off (#247)', async () => {
+    const { FOREIGN_RUNS, foreignRunsState } = await sourcesModule;
+    assert.deepEqual(FOREIGN_RUNS, { id: 'foreign-runs', key: 'save_foreign_runs', label: 'Save prompts from other workflows',
+        description: 'New outputs of workflows without a PromptManager node add their prompt to the library when it is certain.',
+        off: 'Applies while Watch for new outputs is on.' });
+    assert.deepEqual(foreignRunsState({ value: true, monitoring: true }), { disabled: false, checked: true, note: null });
+    assert.deepEqual(foreignRunsState({ value: false, monitoring: true }), { disabled: false, checked: false, note: null });
+    // Monitoring drafted or saved off: nothing is watched, so the row says why it is disabled and keeps its value.
+    assert.deepEqual(foreignRunsState({ value: true, monitoring: false }), { disabled: true, checked: true, note: FOREIGN_RUNS.off });
+    // Not read yet (or the host read failed): disabled without a value, the host rows' own note applies.
+    assert.deepEqual(foreignRunsState({ value: undefined, monitoring: undefined }), { disabled: true, checked: null, note: null });
+    assert.deepEqual(foreignRunsState({ value: undefined, monitoring: false }), { disabled: true, checked: null, note: null });
+    // Saved off while the running monitor still watches until the restart (D128 `stopping`): it can still be set.
+    assert.deepEqual(foreignRunsState({ value: true, monitoring: false, stopping: true }),
+        { disabled: false, checked: true, note: FOREIGN_RUNS.off });
+});

tests/test_api_diagnostics_report.py

@@ -85,6 +85,7 @@ CONFIG = {
         "max_image_age_days",
         "max_concurrent_processing",
         "metadata_extraction_timeout",
+        "save_foreign_runs",
     },
     "integrations": {"lora_manager_enabled", "has_civitai_api_key"},
 }

tests/test_config.py

@@ -59,6 +59,28 @@ class TestGalleryConfig(unittest.TestCase):
         self.assertFalse(GalleryConfig.MONITORING_ENABLED)
         self.assertEqual(GalleryConfig.PROCESSING_DELAY, 5.0)
 
+    def test_save_foreign_runs_defaults_on_and_only_booleans_apply(self):
+        """#247: gallery.monitoring.save_foreign_runs, default true."""
+        self.assertIs(
+            GalleryConfig.get_config()["monitoring"]["save_foreign_runs"], True
+        )
+        GalleryConfig.update_config({"monitoring": {"save_foreign_runs": False}})
+        self.assertIs(GalleryConfig.SAVE_FOREIGN_RUNS, False)
+        for junk in ("true", 1, None):
+            with self.subTest(value=junk):
+                with self.assertLogs("prompt_manager.config", "WARNING") as logs:
+                    GalleryConfig.update_config(
+                        {"monitoring": {"save_foreign_runs": junk}}
+                    )
+                self.assertIn(
+                    "Ignoring gallery monitoring save_foreign_runs: expected true "
+                    "or false",
+                    "\n".join(logs.output),
+                )
+                self.assertIs(GalleryConfig.SAVE_FOREIGN_RUNS, False)
+        GalleryConfig.update_config({"monitoring": {"save_foreign_runs": True}})
+        self.assertIs(GalleryConfig.SAVE_FOREIGN_RUNS, True)
+
     def test_update_tracking(self):
         GalleryConfig.update_config({"tracking": {"prompt_timeout": 300}})
         self.assertEqual(GalleryConfig.PROMPT_TIMEOUT, 300)

tests/test_foreign_save_to_library.py

@@ -0,0 +1,330 @@
+"""#247 PR 2 x #90: the monitor's foreign path never stores switched-off text.
+
+A file whose run's prompt has Save to library off is the opted-out path's,
+not the foreign path's; a foreign graph whose ladder decision is
+``not_saved`` (a switched-off node the #90 walk does not reach) stores
+nothing at settle, at the re-check after the seal, or from recorded history,
+and leaves no wait behind.
+"""
+
+import os
+import unittest
+
+from tests import saver_anchor_graphs as shapes
+from tests import test_run_attribution as attribution
+from tests import test_save_to_library as save_tests
+from tests.test_pair_scan import sampled_graph
+from utils.foreign_runs import ForeignRecheck
+from utils.hashing import generate_prompt_hash
+from py.config import GalleryConfig  # after tests: it stubs the host
+
+LOGGER = "prompt_manager.image_monitor"
+OFF_TEXT = "SECRET OFF TEXT"
+SKIPPED = "Save to library is off for this run's prompt, skipping image"
+
+# The class dict, not the class: a TestCase bound here would run again
+Fixture = vars(attribution.TestMonitorForeignRuns)
+
+
+def off_behind(middle="MyNode", key="value"):
+    """A switched-off PromptManagerText reaching a core CLIPTextEncode through
+    ``middle``'s ``key`` input: the #90 walk does not follow ``key``."""
+    graph = save_tests.run_graph()
+    graph["1"] = {
+        "class_type": "PromptManagerText",
+        "inputs": {"text": OFF_TEXT, "save_to_library": False},
+    }
+    graph["7"] = {"class_type": middle, "inputs": {key: ["1", 0]}}
+    graph["8"] = {
+        "class_type": "CLIPTextEncode",
+        "inputs": {"text": ["7", 0], "clip": ["4", 1]},
+    }
+    graph["3"]["inputs"]["positive"] = ["8", 0]
+    return graph
+
+
+def off_text_unwired():
+    """Only history can decide; the legacy answer is a dangling switched-off
+    node's own text (the ladder says ``not_saved`` once it may ask history)."""
+    graph = sampled_graph("unused", "blurry")
+    graph["1"]["inputs"]["text"] = ["30", 0]
+    graph["30"] = {"class_type": "VendorPromptMaker", "inputs": {"prompt": OFF_TEXT}}
+    graph["40"] = {
+        "class_type": "PromptManagerText",
+        "inputs": {"text": OFF_TEXT, "save_to_library": False},
+    }
+    return graph
+
+
+class TestForeignPathKeepsSaveToLibrary(attribution.RecordedRuns, unittest.TestCase):
+    setUp = Fixture["setUp"]
+    close_databases = Fixture["close_databases"]
+    wait_for = Fixture["wait_for"]
+    image = Fixture["image"]
+    linked_text = Fixture["linked_text"]
+    key = Fixture["key"]
+    prompt_count = Fixture["prompt_count"]
+    settled = Fixture["settled"]
+
+    def reference(self, path):
+        return {
+            "folder_type": "output",
+            "subfolder": "dataset",
+            "filename": os.path.basename(path),
+        }
+
+    def seal(self, path, positives=(), nodes=("9",)):
+        """Record one run reporting ``path`` (a new run id for every call)."""
+        self.runs = getattr(self, "runs", -1) + 1
+        self.record_run(
+            self.db,
+            self.output,
+            [("dataset", os.path.basename(path))],
+            list(positives),
+            item=self.runs,
+            nodes=list(nodes),
+        )
+
+    def recheck(self, path):
+        """The re-check the seal schedules, run through the monitor's entry."""
+        with self.assertLogs(LOGGER, "INFO") as logs:
+            self.handler.process_new_image(path, ForeignRecheck(self.reference(path)))
+        return "\n".join(logs.output)
+
+    def assert_nothing_stored(self, path):
+        self.assertIsNone(self.linked_text(path))
+        self.assertEqual(self.prompt_count(), 0)
+        self.assertIsNone(self.db.get_prompt_by_hash(generate_prompt_hash(OFF_TEXT)))
+        self.assertEqual(self.handler.awaiting_foreign(), [])
+        self.assertEqual(list(self.handler._foreign_due), [])
+
+    def test_q1_the_only_prompt_manager_node_off_is_opted_out_not_foreign(self):
+        graph = save_tests.run_graph(save=False)
+        # The switched-off positive is the only PromptManager node
+        graph["2"] = {"class_type": "CLIPTextEncode", "inputs": {"text": "blurry"}}
+        sampler_less = {
+            "1": {
+                "class_type": "PromptManagerText",
+                "inputs": {"text": OFF_TEXT, "save_to_library": False},
+            }
+        }
+        for name, graph in (("sampled", graph), ("sampler-less", sampler_less)):
+            with self.subTest(name=name):
+                self.metadata = {"prompt": graph}
+                state, _ = self.handler._find_prompt_from_metadata(self.metadata)
+                self.assertEqual(state, "opted_out")
+                path = self.image(f"IMG_GEN_{name}_00001_.png")
+                with self.assertLogs(LOGGER, "INFO") as logs:
+                    self.handler.process_new_image(path)
+                self.assertIn(SKIPPED, "\n".join(logs.output))
+                self.seal(path)
+                self.assertTrue(self.settled())
+                self.assertIn(SKIPPED, self.recheck(path))
+                self.assert_nothing_stored(path)
+
+    def test_q2_an_off_text_node_behind_a_pass_through_stores_nothing(self):
+        for middle, key in (("MyNode", "value"), ("easy showAnything", "anything")):
+            with self.subTest(middle=middle):
+                self.metadata = {"prompt": off_behind(middle, key)}
+                state, _ = self.handler._find_prompt_from_metadata(self.metadata)
+                self.assertEqual(state, "foreign")
+                path = self.image(f"IMG_GEN_{key}_00001_.png")
+                with self.assertLogs(LOGGER, "INFO") as logs:
+                    self.handler.process_new_image(path)
+                self.assertIn(SKIPPED, "\n".join(logs.output))
+                self.assert_nothing_stored(path)
+                self.seal(path)
+                self.assertTrue(self.settled())
+                self.assertIn(SKIPPED, self.recheck(path))
+                self.assert_nothing_stored(path)
+
+    def test_q2_a_wired_switch_behind_a_pass_through_stores_nothing(self):
+        graph = off_behind()
+        graph["1"]["inputs"]["save_to_library"] = ["20", 0]
+        graph["20"] = {"class_type": "SomeBoolNode", "inputs": {}}
+        self.metadata = {"prompt": graph}
+        path = self.image("IMG_GEN_wired_00001_.png")
+        wired = "Save to library is wired from a node the image cannot show"
+        with self.assertLogs(LOGGER, "INFO") as logs:
+            self.handler.process_new_image(path)
+        self.assertIn(wired, "\n".join(logs.output))
+        self.seal(path)
+        self.assertTrue(self.settled())
+        self.assertIn(wired, self.recheck(path))
+        self.assert_nothing_stored(path)
+
+    def test_q3_an_undecided_not_saved_waits_for_the_record_then_skips(self):
+        """No record yet: a not_saved of the undecided step is not final (a
+        recorded saver may resolve another branch), so the file waits for its
+        run; the decision with the record stores nothing and logs the skip."""
+        wired = off_text_unwired()
+        wired["40"]["inputs"]["save_to_library"] = ["20", 0]
+        wired["20"] = {"class_type": "SomeBoolNode", "inputs": {}}
+        cases = (
+            ("off", off_text_unwired(), SKIPPED),
+            ("wired", wired, "Save to library is wired from a node"),
+        )
+        for name, graph, line in cases:
+            with self.subTest(name=name):
+                self.metadata = {"prompt": graph}
+                path = self.image(f"vendor_{name}_00001_.png")
+                self.handler.process_new_image(path)
+                self.assertIsNone(self.linked_text(path))
+                self.assertEqual(self.handler.awaiting_foreign(), [self.key(path)])
+                with self.assertLogs(LOGGER, "INFO") as logs:
+                    self.seal(path)
+                    self.assertTrue(
+                        self.wait_for(lambda: not self.handler.awaiting_foreign())
+                    )
+                    self.assertTrue(self.settled())
+                self.assertIn(line, "\n".join(logs.output))
+                self.assert_nothing_stored(path)
+
+    def test_q3_an_early_not_saved_never_waits_even_with_a_saver_branch(self):
+        """Review: a switched-off node feeding one branch's positive is the
+        ladder's early, final not_saved (PR 1's #90 rule covers every
+        sampler's positive), so the file never waits and nothing is stored,
+        not even the plain branch the recorded saver names."""
+        graph = shapes.shared_prefix()
+        graph["20"]["inputs"]["text"] = ["50", 0]
+        graph["50"] = {
+            "class_type": "PromptManagerText",
+            "inputs": {"text": OFF_TEXT, "save_to_library": False},
+        }
+        self.metadata = {"prompt": graph}
+        path = self.image("IMG_GEN_00001_.png")
+        with self.assertLogs(LOGGER, "INFO") as logs:
+            self.handler.process_new_image(path)
+        self.assertIn(SKIPPED, "\n".join(logs.output))
+        self.assert_nothing_stored(path)
+        self.seal(path, nodes=("13",))
+        self.assertTrue(self.settled())
+        self.assertIn(SKIPPED, self.recheck(path))
+        self.assert_nothing_stored(path)
+        self.assertIsNone(
+            self.db.get_prompt_by_hash(generate_prompt_hash(shapes.BRANCH_TEXTS[0]))
+        )
+
+    def test_q3_an_off_node_in_the_recorded_saver_branch_is_never_foreign(self):
+        """PR 3's parity case: a switched-off node feeds the very branch the
+        recorded saver names (saver "23" is branch 2). A literal switch is the
+        opted-out path's; a wired one is the node's own run's (#90), never the
+        foreign path's. Nothing is stored at settle or after the run, and no
+        file waits for a foreign re-check."""
+        cases = (
+            ("off", False, "opted_out", SKIPPED),
+            ("wired", ["52", 0], "unknown", "No certain prompt for"),
+        )
+        for name, save, state, line in cases:
+            with self.subTest(name=name):
+                graph = shapes.shared_prefix()
+                graph["50"] = {
+                    "class_type": "PromptManagerText",
+                    "inputs": {"text": OFF_TEXT, "save_to_library": save},
+                }
+                graph["20"]["inputs"]["text"] = ["50", 0]
+                if isinstance(save, list):
+                    graph["52"] = {"class_type": "SomeBoolNode", "inputs": {}}
+                self.metadata = {"prompt": graph}
+                found, _ = self.handler._find_prompt_from_metadata(self.metadata)
+                self.assertEqual(found, state)
+                path = self.image(f"IMG_GEN_{name}_00001_.png")
+                with self.assertLogs(LOGGER, "INFO") as logs:
+                    self.handler.process_new_image(path)
+                self.assertIn(line, "\n".join(logs.output))
+                self.assert_nothing_stored(path)
+                self.seal(path, positives=[OFF_TEXT], nodes=("23",))
+                self.assertTrue(self.settled())
+                self.handler.process_new_image(path)
+                self.assert_nothing_stored(path)
+                self.assertIsNone(
+                    self.db.get_prompt_by_hash(
+                        generate_prompt_hash(shapes.BRANCH_TEXTS[1])
+                    )
+                )
+
+    def test_q4_an_off_node_on_the_negative_only_still_links_the_positive(self):
+        positive = "a lighthouse on a cliff at dusk"
+        plain = sampled_graph(positive, "unused")
+        plain["2"] = {
+            "class_type": "PromptManager",
+            "inputs": {"text": OFF_TEXT, "save_to_library": False},
+        }
+        behind = sampled_graph(positive, "unused")
+        behind["2"]["inputs"]["text"] = ["7", 0]
+        behind["7"] = {
+            "class_type": "PromptManagerText",
+            "inputs": {"text": OFF_TEXT, "save_to_library": False},
+        }
+        for name, graph in (("direct", plain), ("behind an encoder", behind)):
+            with self.subTest(name=name):
+                self.metadata = {"prompt": graph}
+                state, _ = self.handler._find_prompt_from_metadata(self.metadata)
+                self.assertEqual(state, "foreign")
+                path = self.image(f"plain_{len(name)}_00001_.png")
+                self.handler.process_new_image(path)
+                self.assertEqual(self.linked_text(path), positive)
+                self.assertIsNone(
+                    self.db.get_prompt_by_hash(generate_prompt_hash(OFF_TEXT))
+                )
+                pairs = self.db.list_prompt_pairs()["pairs"]
+                self.assertEqual(pairs, [])
+                self.assertEqual(self.handler.awaiting_foreign(), [])
+
+    def test_q5_recorded_history_never_brings_back_switched_off_text(self):
+        """Even a recorded run whose contexts hold the off text (an older
+        history row) cannot name it, at settle or at the re-check."""
+        cases = {"pass-through": off_behind(), "dangling": off_text_unwired()}
+        for name, graph in cases.items():
+            with self.subTest(name=name):
+                self.metadata = {"prompt": graph}
+                path = self.image(f"IMG_GEN_{len(name)}_00001_.png")
+                self.seal(path, positives=[OFF_TEXT], nodes=("9",))
+                self.handler.process_new_image(path)
+                self.assert_nothing_stored(path)
+                self.assertIn(SKIPPED, self.recheck(path))
+                self.assert_nothing_stored(path)
+
+    def test_q6_with_foreign_saving_off_save_to_library_is_unchanged(self):
+        GalleryConfig.SAVE_FOREIGN_RUNS = False
+        opted_out = save_tests.run_graph(save=False)
+        self.metadata = {"prompt": opted_out}
+        skipped = self.image("IMG_GEN_off_00001_.png")
+        with self.assertLogs(LOGGER, "INFO") as logs:
+            self.handler.process_new_image(skipped)
+        self.assertIn(SKIPPED, "\n".join(logs.output))
+        self.metadata = {"prompt": off_behind()}
+        foreign = self.image("IMG_GEN_foreign_00001_.png")
+        with self.assertLogs(LOGGER, "INFO") as logs:
+            self.handler.process_new_image(foreign)
+        self.assertIn("No PromptManager prompt in workflow", "\n".join(logs.output))
+        for path in (skipped, foreign):
+            self.assert_nothing_stored(path)
+        # A switched-on PromptManager positive still links by its metadata
+        stored = self.db.save_prompt(
+            text=save_tests.POSITIVE,
+            prompt_hash=generate_prompt_hash(save_tests.POSITIVE),
+        )
+        self.metadata = {"prompt": save_tests.run_graph(save=True)}
+        on = self.image("IMG_GEN_on_00001_.png")
+        self.handler.process_new_image(on)
+        self.assertEqual(self.linked_text(on), save_tests.POSITIVE)
+        self.assertEqual(
+            self.db.get_prompt_by_hash(generate_prompt_hash(save_tests.POSITIVE))["id"],
+            stored,
+        )
+
+    def test_the_shared_prefix_dataset_still_links_after_the_seal(self):
+        """Regression guard (passes before the fix): the wait still works."""
+        self.metadata = {"prompt": shapes.shared_prefix()}
+        path = self.image("IMG_GEN_00001_.png")
+        self.handler.process_new_image(path)
+        self.assertEqual(self.handler.awaiting_foreign(), [self.key(path)])
+        self.seal(path, nodes=("13",))
+        self.assertTrue(self.wait_for(lambda: self.linked_text(path)))
+        self.assertTrue(self.settled())
+        self.assertEqual(self.linked_text(path), shapes.BRANCH_TEXTS[0])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_image_linking.py

@@ -172,17 +172,40 @@ class TestForeignWorkflows(LinkingTestCase):
         self.tracker.set_current_prompt(text, {"prompt_id": prompt_id})
         return prompt_id
 
+    def _linked_text(self):
+        row = (
+            self.db.model.get_connection()
+            .execute(
+                "SELECT p.text FROM generated_images g "
+                "JOIN prompts p ON p.id = g.prompt_id WHERE g.image_path = ?",
+                (self.image,),
+            )
+            .fetchone()
+        )
+        return row[0] if row else None
+
     def test_foreign_image_leaves_the_batch_queue_aligned(self):
+        """#247: it links to its own prompt and never takes a queue entry."""
         first = self._queue_batch_item("item one")
         self.assertIsNone(self._process({"prompt": foreign_graph()}))
         self.assertEqual(self.linked, [])
         self.assertEqual(self.tracker.pop_next_prompt()["id"], first)
         self.assertIsNone(self.tracker.pop_next_prompt())
+        self.assertEqual(self._linked_text(), "a cat")
+        prompt = self.db.get_prompt_by_hash(generate_prompt_hash("a cat"))
+        self.assertEqual(prompt["run_count"] or 0, 0)
 
-    def test_foreign_image_is_not_linked_by_the_later_fallbacks_either(self):
+    def test_a_foreign_image_is_linked_only_by_the_ladder(self):
         self._queue_batch_item("item one")  # live tracker and recent prompt exist
+        untraceable = foreign_graph()
+        untraceable["9"]["inputs"]["text"] = ["30", 0]
+        untraceable["30"] = {"class_type": "VendorPromptMaker", "inputs": {}}
+        self.assertIsNone(self._process({"prompt": untraceable}))
+        self.assertEqual(self.linked, [])
+        self.assertIsNone(self._linked_text())
         self.assertIsNone(self._process({"prompt": foreign_graph()}))
         self.assertEqual(self.linked, [])
+        self.assertEqual(self._linked_text(), "a cat")
 
     def test_negative_only_workflow_is_foreign(self):
         self._save("blurry")

tests/test_image_monitor_singleton.py

@@ -597,6 +597,11 @@ class TestHandlerScheduling(unittest.TestCase):
             self.handler._schedule_recheck(path)
         self.assertIsNone(self.handler._worker)
         self.assertEqual(list(self.handler._history_due), [])
+        # #247: a foreign file's sealed re-check is dropped the same way
+        self.handler._await_foreign(path)
+        self.assertEqual(self.handler.awaiting_foreign(), [])
+        self.handler._schedule_recheck(path, {"filename": "late.png"})
+        self.assertEqual(list(self.handler._foreign_due), [])
         self.assertEqual(self._live_workers() - before, set())
 
     def test_a_sealed_run_starts_a_worker_on_an_open_handler(self):

tests/test_run_attribution.py

@@ -33,6 +33,7 @@ from utils.execution_outputs import (
     snapshot_output_roots,
 )
 from utils.hashing import generate_prompt_hash
+from py.config import GalleryConfig
 from utils.image_monitor import ImageGenerationHandler
 
 STYLE = "windows" if os.name == "nt" else "posix"
@@ -771,6 +772,470 @@ class TestMonitorHistoryRecheck(RecordedRuns, unittest.TestCase):
         run_attribution.notify_outputs_sealed("run", [])  # no dead callback raises
 
 
+class TestMonitorForeignRuns(RecordedRuns, unittest.TestCase):
+    """#247: outputs of workflows without a PromptManager node, in the monitor."""
+
+    def setUp(self):
+        history_tests.HistoryDatabaseTests.setUp(self)
+        self.db = history_tests.HistoryDatabaseTests.database(self)
+        self.output = str(Path(self.temp.name) / "output")
+        os.makedirs(os.path.join(self.output, "dataset"))
+        self.tracker = make_tracker(self.db)
+        self.handler = ImageGenerationHandler(self.db, self.tracker)
+        self.handler.processing_delay = 0
+        self.handler.settle_interval = 0.01
+        self.addCleanup(close_handler, self, self.handler)
+        self.metadata = {"prompt": shapes.shared_prefix()}
+        saved = GalleryConfig.SAVE_FOREIGN_RUNS
+        self.addCleanup(setattr, GalleryConfig, "SAVE_FOREIGN_RUNS", saved)
+        GalleryConfig.SAVE_FOREIGN_RUNS = True
+        patches = (
+            mock.patch(
+                "utils.image_monitor.host_output_directory", return_value=self.output
+            ),
+            mock.patch(
+                "database.run_attribution.host_output_directory",
+                return_value=self.output,
+            ),
+            mock.patch.object(
+                self.handler.metadata_extractor,
+                "extract_metadata",
+                side_effect=lambda path: self.metadata,
+            ),
+        )
+        for patch in patches:
+            patch.start()
+            self.addCleanup(patch.stop)
+
+    close_databases = history_tests.HistoryDatabaseTests.close_databases
+    wait_for = TestMonitorHistoryRecheck.wait_for
+
+    def image(self, name, subfolder="dataset"):
+        path = os.path.join(self.output, subfolder, name)
+        Path(path).write_bytes(b"png")
+        return path
+
+    def linked_text(self, path):
+        row = (
+            self.db.model.get_connection()
+            .execute(
+                "SELECT p.text FROM generated_images g "
+                "JOIN prompts p ON p.id = g.prompt_id WHERE g.image_path = ?",
+                (str(path),),
+            )
+            .fetchone()
+        )
+        return row[0] if row else None
+
+    def key(self, path):
+        return self.handler._awaiting_key(path)
+
+    def prompt_count(self):
+        connection = self.db.model.get_connection()
+        return connection.execute("SELECT COUNT(*) FROM prompts").fetchone()[0]
+
+    def settled(self):
+        return self.wait_for(lambda: self.handler.pending_count() == 0)
+
+    def test_each_dataset_file_links_to_its_branch_once_the_run_ends(self):
+        """CUJ 1: unlinked while the run is open, its own branch after the seal."""
+        paths = [self.image(f"IMG_GEN_0000{n}_.png") for n in (1, 2)]
+        for path in paths:
+            self.handler.process_new_image(path)
+        self.assertEqual([self.linked_text(path) for path in paths], [None, None])
+        self.assertEqual(
+            self.handler.awaiting_foreign(), [self.key(path) for path in paths]
+        )
+        self.assertEqual(self.handler.awaiting_history(), [])
+        self.record_run(
+            self.db,
+            self.output,
+            [("dataset", os.path.basename(path)) for path in paths],
+            [],
+            nodes=["13", "23"],
+        )
+        self.assertTrue(self.wait_for(lambda: all(map(self.linked_text, paths))))
+        self.assertTrue(self.settled())
+        self.assertEqual(
+            [self.linked_text(path) for path in paths], list(shapes.BRANCH_TEXTS[:2])
+        )
+        self.assertEqual(self.handler.awaiting_foreign(), [])
+        prompt = self.db.get_prompt_by_hash(
+            generate_prompt_hash(shapes.BRANCH_TEXTS[0])
+        )
+        self.assertEqual(prompt["category"], "scanned")
+        self.assertIn("Saved from a workflow run: IMG_GEN_00001_.png", prompt["notes"])
+
+    def test_a_run_sealed_before_the_file_settles_links_it_at_once(self):
+        path = self.image("IMG_GEN_00003_.png")
+        self.record_run(
+            self.db, self.output, [("dataset", "IMG_GEN_00003_.png")], [], nodes=["33"]
+        )
+        self.handler.process_new_image(path)
+        self.assertEqual(self.linked_text(path), shapes.BRANCH_TEXTS[2])
+        self.assertEqual(self.handler.awaiting_foreign(), [])
+
+    def test_a_name_two_runs_reported_stays_unlinked(self):
+        """CUJ 3: the file was overwritten; no run proves its saver."""
+        path = self.image("IMG_GEN_00004_.png")
+        reported = [("dataset", "IMG_GEN_00004_.png")]
+        self.record_run(self.db, self.output, reported, [], nodes=["13"])
+        self.record_run(self.db, self.output, reported, [], item=1, nodes=["23"])
+        self.handler.process_new_image(path)
+        self.assertIsNone(self.linked_text(path))
+        reference = {"folder_type": "output", "subfolder": "dataset"}
+        reference["filename"] = "IMG_GEN_00004_.png"
+        with self.assertLogs("prompt_manager.image_monitor", "INFO") as logs:
+            self.handler._recheck_foreign(path, reference)
+        self.assertIn(
+            "No certain prompt for IMG_GEN_00004_.png after its run ended; "
+            "left unlinked",
+            "\n".join(logs.output),
+        )
+        self.assertIsNone(self.linked_text(path))
+        self.assertEqual(self.prompt_count(), 0)
+
+    def test_the_setting_off_stores_nothing_at_settle_or_after_the_run(self):
+        """CUJ 4: read for every file, so it applies without a restart."""
+        GalleryConfig.SAVE_FOREIGN_RUNS = False
+        skipped = self.image("IMG_GEN_00005_.png")
+        with self.assertLogs("prompt_manager.image_monitor", "INFO") as logs:
+            self.handler.process_new_image(skipped)
+        self.assertIn(
+            "No PromptManager prompt in workflow, skipping image",
+            "\n".join(logs.output),
+        )
+        self.assertEqual(self.handler.awaiting_foreign(), [])
+        GalleryConfig.SAVE_FOREIGN_RUNS = True
+        waiting = self.image("IMG_GEN_00006_.png")
+        self.handler.process_new_image(waiting)
+        self.assertEqual(self.handler.awaiting_foreign(), [self.key(waiting)])
+        GalleryConfig.SAVE_FOREIGN_RUNS = False
+        self.record_run(
+            self.db, self.output, [("dataset", "IMG_GEN_00006_.png")], [], nodes=["13"]
+        )
+        self.assertTrue(self.wait_for(lambda: not self.handler.awaiting_foreign()))
+        self.assertTrue(self.settled())
+        self.assertEqual(
+            [self.linked_text(p) for p in (skipped, waiting)], [None, None]
+        )
+        self.assertEqual(self.prompt_count(), 0)
+
+    def test_a_plain_foreign_workflow_links_at_settle_and_counts_no_run(self):
+        """CUJ 2: the default profile decides; the tracker queue is untouched."""
+        text = "a lighthouse on a cliff at dusk"
+        self.metadata = {"prompt": sampled_graph(text, "blurry")}
+        queued = self.db.save_prompt(
+            text="queued item", prompt_hash=generate_prompt_hash("queued item")
+        )
+        self.tracker.set_current_prompt("queued item", {"prompt_id": queued})
+        first, second = self.image("plain_00001_.png"), self.image("plain_00002_.png")
+        self.handler.process_new_image(first)
+        self.handler.process_new_image(second)
+        self.assertEqual(
+            [self.linked_text(first), self.linked_text(second)], [text, text]
+        )
+        prompt = self.db.get_prompt_by_hash(generate_prompt_hash(text))
+        self.assertEqual(prompt["run_count"] or 0, 0)
+        self.assertIsNone(prompt.get("last_used_at"))
+        self.assertEqual(self.tracker.pop_next_prompt()["id"], queued)
+        self.assertEqual(self.handler.awaiting_foreign(), [])
+
+    def test_the_sealed_reference_spells_the_file_as_history_recorded_it(self):
+        """Windows: the watcher's spelling differs in case from the run's report."""
+        Path(self.output, "Dataset").mkdir()
+        recorded = self.image("IMG_GEN_00007_.png", subfolder="Dataset")
+        seen = os.path.join(self.output, "dataset", "img_gen_00007_.png")
+        Path(seen).write_bytes(b"png")
+        folded = staticmethod(lambda path: os.path.realpath(path).lower())
+        with mock.patch.object(ImageGenerationHandler, "_awaiting_key", folded):
+            self.handler.process_new_image(seen)
+            self.assertIsNone(self.linked_text(seen))
+            self.record_run(
+                self.db,
+                self.output,
+                [("Dataset", os.path.basename(recorded))],
+                [],
+                nodes=["43"],
+            )
+            self.assertTrue(self.wait_for(lambda: self.linked_text(seen)))
+            self.assertTrue(self.settled())
+        self.assertEqual(self.linked_text(seen), shapes.BRANCH_TEXTS[3])
+
+    def test_a_full_foreign_set_never_evicts_a_waiting_keyless_file(self):
+        from utils import image_monitor
+
+        keyless = self.image("bare_00001_.png", subfolder="")
+        with mock.patch.object(
+            self.handler.metadata_extractor, "extract_metadata", return_value=None
+        ):
+            self.handler.process_new_image(keyless)
+        paths = [self.image(f"IMG_GEN_1000{n}_.png") for n in range(3)]
+        with (
+            mock.patch.object(image_monitor, "MAX_AWAITING_HISTORY", 2),
+            self.assertLogs("prompt_manager.image_monitor", "INFO") as logs,
+        ):
+            for path in paths:
+                self.handler.process_new_image(path)
+            self.assertEqual(
+                self.handler.awaiting_foreign(), [self.key(p) for p in paths[1:]]
+            )
+        self.assertEqual(self.handler.awaiting_history(), [self.key(keyless)])
+        self.assertIn(
+            "Too many files wait for their run; IMG_GEN_10000_.png left unlinked",
+            "\n".join(logs.output),
+        )
+
+    def test_an_expired_foreign_entry_is_logged_and_left_unlinked(self):
+        from utils import image_monitor
+
+        path = self.image("IMG_GEN_00008_.png")
+        self.handler.process_new_image(path)
+        later = time.monotonic() + image_monitor.AWAIT_HISTORY_SECONDS + 1
+        with (
+            mock.patch.object(image_monitor.time, "monotonic", return_value=later),
+            self.assertLogs("prompt_manager.image_monitor", "INFO") as logs,
+        ):
+            self.assertEqual(self.handler.awaiting_foreign(), [])
+        self.assertIn(
+            "No recorded run for IMG_GEN_00008_.png; left unlinked",
+            "\n".join(logs.output),
+        )
+        self.assertIsNone(self.linked_text(path))
+
+    def test_a_seal_while_the_pending_set_is_full_is_rechecked_later(self):
+        from utils import image_monitor
+
+        path = self.image("IMG_GEN_00009_.png")
+        self.handler.process_new_image(path)
+        blocker = self.image("blocker.png", subfolder="")
+        release = threading.Event()
+        original = self.handler.process_new_image
+
+        def slow(image_path, prompt_snapshot=None):
+            if image_path == blocker:
+                release.wait(5)
+                return None
+            return original(image_path, prompt_snapshot=prompt_snapshot)
+
+        self.handler.process_new_image = slow
+        with mock.patch.object(image_monitor, "MAX_PENDING_IMAGES", 1):
+            self.assertTrue(self.handler.schedule(blocker))
+            self.assertTrue(self.wait_for(lambda: self.handler.pending_count() == 1))
+            self.record_run(
+                self.db,
+                self.output,
+                [("dataset", "IMG_GEN_00009_.png")],
+                [],
+                nodes=["13"],
+            )
+            self.assertEqual(list(self.handler._foreign_due), [path])
+            release.set()
+            self.assertTrue(self.wait_for(lambda: self.linked_text(path)))
+        self.assertTrue(self.settled())
+        self.assertEqual(self.linked_text(path), shapes.BRANCH_TEXTS[0])
+
+    def test_shutdown_forgets_waiting_and_due_foreign_files(self):
+        path = self.image("IMG_GEN_00010_.png")
+        self.handler.process_new_image(path)
+        self.handler._foreign_due[path] = {"filename": "IMG_GEN_00010_.png"}
+        self.handler.shutdown()
+        self.assertEqual(self.handler.awaiting_foreign(), [])
+        self.assertEqual(list(self.handler._foreign_due), [])
+
+    def link_count(self, path):
+        connection = self.db.model.get_connection()
+        return connection.execute(
+            "SELECT COUNT(*) FROM generated_images WHERE image_path = ?", (str(path),)
+        ).fetchone()[0]
+
+    def test_a_file_linked_to_another_prompt_never_gets_a_second_link(self):
+        """One prompt per file: the pair path and the positive path both refuse."""
+        other = self.db.save_prompt(
+            "an earlier prompt", generate_prompt_hash("an earlier prompt")
+        )
+        plain = self.image("plain_00004_.png")
+        self.db.link_image_to_prompt(other, plain)
+        self.metadata = {"prompt": sampled_graph("a quiet harbor", "blurry")}
+        with self.assertLogs("prompt_manager.image_monitor", "INFO") as logs:
+            self.handler.process_new_image(plain)
+        self.assertIn("Foreign run: conflict", "\n".join(logs.output))
+        dataset = self.image("IMG_GEN_00011_.png")
+        self.db.link_image_to_prompt(other, dataset)
+        self.metadata = {"prompt": shapes.shared_prefix()}
+        self.record_run(
+            self.db, self.output, [("dataset", "IMG_GEN_00011_.png")], [], nodes=["13"]
+        )
+        self.handler.process_new_image(dataset)
+        for path in (plain, dataset):
+            self.assertEqual(self.linked_text(path), "an earlier prompt")
+            self.assertEqual(self.link_count(path), 1)
+        for text in ("a quiet harbor", shapes.BRANCH_TEXTS[0]):
+            self.assertIsNone(self.db.get_prompt_by_hash(generate_prompt_hash(text)))
+
+    def test_the_file_size_is_read_only_when_the_ladder_needs_it(self):
+        """PR 1's lazy size: a foreign file decided without a store reads none."""
+        # Foreign (a core encoder only) and no API graph: never, at step 3
+        encoder = {"type": "CLIPTextEncode", "inputs": [], "widgets_values": ["x"]}
+        self.metadata = {"text_encoder_nodes": [encoder]}
+        path = self.image("bare_00002_.png", subfolder="")
+        with mock.patch("utils.foreign_runs.os.path.getsize") as size:
+            self.handler.process_new_image(path)
+        size.assert_not_called()
+
+    def untraceable(self):
+        """Unresolved and only the legacy parser names a text (an unknown node)."""
+        graph = sampled_graph("unused", "blurry")
+        graph["1"]["inputs"]["text"] = ["30", 0]
+        graph["30"] = {
+            "class_type": "VendorPromptMaker",
+            "inputs": {"prompt": "a misty valley at dawn"},
+        }
+        return {"prompt": graph}
+
+    def test_no_record_at_settle_keeps_the_file_waiting_for_the_legacy_floor(self):
+        """Review: a lookup that found no record is not the run's answer yet."""
+        self.metadata = self.untraceable()
+        path = self.image("vendor_00001_.png")
+        self.handler.process_new_image(path)
+        self.assertIsNone(self.linked_text(path))
+        self.assertEqual(self.handler.awaiting_foreign(), [self.key(path)])
+        self.record_run(self.db, self.output, [("dataset", "vendor_00001_.png")], [])
+        self.assertTrue(self.wait_for(lambda: self.linked_text(path)))
+        self.assertTrue(self.settled())
+        self.assertEqual(self.linked_text(path), "a misty valley at dawn")
+
+    def test_a_record_found_at_settle_without_a_certain_prompt_ends_the_wait(self):
+        """Review: sealed before settle, still ambiguous: no seal is coming."""
+        path = self.image("IMG_GEN_00012_.png")
+        # One run reported it from two savers: a record, but no saver
+        reported = [("dataset", "IMG_GEN_00012_.png")] * 2
+        self.record_run(self.db, self.output, reported, [], nodes=["13", "23"])
+        with self.assertLogs("prompt_manager.image_monitor", "INFO") as logs:
+            self.handler.process_new_image(path)
+        self.assertIn(
+            "No certain prompt for IMG_GEN_00012_.png; left unlinked",
+            "\n".join(logs.output),
+        )
+        self.assertEqual(self.handler.awaiting_foreign(), [])
+        self.assertIsNone(self.linked_text(path))
+
+    def test_monitoring_saved_off_stops_foreign_writes_at_settle_and_recheck(self):
+        """Review: the observer runs until a restart; nothing is written."""
+        enabled = GalleryConfig.MONITORING_ENABLED
+        self.addCleanup(setattr, GalleryConfig, "MONITORING_ENABLED", enabled)
+        GalleryConfig.MONITORING_ENABLED = False
+        self.metadata = {"prompt": sampled_graph("a quiet harbor", "blurry")}
+        plain = self.image("plain_00005_.png")
+        self.handler.process_new_image(plain)
+        self.assertIsNone(self.linked_text(plain))
+        GalleryConfig.MONITORING_ENABLED = True
+        self.metadata = {"prompt": shapes.shared_prefix()}
+        waiting = self.image("IMG_GEN_00013_.png")
+        self.handler.process_new_image(waiting)
+        GalleryConfig.MONITORING_ENABLED = False
+        reference = {"folder_type": "output", "subfolder": "dataset"}
+        reference["filename"] = "IMG_GEN_00013_.png"
+        self.record_run(
+            self.db, self.output, [("dataset", reference["filename"])], [], nodes=["13"]
+        )
+        self.assertTrue(self.wait_for(lambda: not self.handler.awaiting_foreign()))
+        self.assertTrue(self.settled())
+        self.handler._recheck_foreign(waiting, reference)
+        self.assertIsNone(self.linked_text(waiting))
+        self.assertEqual(self.prompt_count(), 0)
+
+    def test_the_setting_is_read_again_just_before_the_store(self):
+        self.metadata = {"prompt": sampled_graph("a quiet harbor", "blurry")}
+        path = self.image("plain_00006_.png")
+        real = __import__("utils.foreign_runs", fromlist=["x"]).decide_foreign
+
+        def decide_then_switch_off(*args, **kwargs):
+            decision = real(*args, **kwargs)
+            GalleryConfig.SAVE_FOREIGN_RUNS = False
+            return decision
+
+        with mock.patch(
+            "utils.image_monitor.decide_foreign", side_effect=decide_then_switch_off
+        ):
+            self.handler.process_new_image(path)
+        self.assertIsNone(self.linked_text(path))
+        self.assertEqual(self.prompt_count(), 0)
+
+    def test_a_seal_between_registering_and_the_settle_lookup_links_once(self):
+        """Regression guard (passes before the review fixes): one link only."""
+        path = self.image("IMG_GEN_00014_.png")
+        real = __import__("utils.foreign_runs", fromlist=["x"]).history_lookup
+
+        def sealed_first(db, image_path, reference=None):
+            lookup = real(db, image_path, reference)
+
+            def run_then_look():
+                if reference is None:  # the settle lookup: the run seals now
+                    self.record_run(
+                        self.db,
+                        self.output,
+                        [("dataset", "IMG_GEN_00014_.png")],
+                        [],
+                        nodes=["13"],
+                    )
+                return lookup()
+
+            return run_then_look
+
+        with mock.patch("utils.image_monitor.history_lookup", side_effect=sealed_first):
+            self.assertTrue(self.handler.schedule(path))
+            self.assertTrue(self.wait_for(lambda: self.linked_text(path)))
+            self.assertTrue(self.settled())
+        self.assertEqual(self.linked_text(path), shapes.BRANCH_TEXTS[0])
+        self.assertEqual(self.link_count(path), 1)
+        self.assertEqual(list(self.handler._foreign_due), [])
+
+    def test_a_drained_recheck_never_drops_a_newer_reference(self):
+        path = self.image("IMG_GEN_00015_.png")
+        older, newer = {"filename": "old"}, {"filename": "new"}
+        self.handler._foreign_due[path] = older
+        real = self.handler.schedule
+
+        def schedule(image_path, snapshot=None):
+            # The run is sealed again while the older one is scheduled
+            self.handler._foreign_due[image_path] = newer
+            return real(image_path, snapshot)
+
+        self.handler.schedule = schedule
+        with mock.patch.object(self.handler, "process_new_image"):
+            self.handler._drain_history_due()
+        self.assertIs(self.handler._foreign_due[path], newer)
+
+    def test_an_evicted_due_recheck_is_logged(self):
+        from utils import image_monitor
+
+        self.handler.schedule = lambda *args, **kwargs: False
+        with (
+            mock.patch.object(image_monitor, "MAX_AWAITING_HISTORY", 1),
+            self.assertLogs("prompt_manager.image_monitor", "INFO") as logs,
+        ):
+            for name in ("a_00001_.png", "b_00001_.png"):
+                self.handler._schedule_recheck(self.image(name), {"filename": name})
+        self.assertIn(
+            "Too many files wait for their run; a_00001_.png left unlinked",
+            "\n".join(logs.output),
+        )
+
+    def test_a_failed_store_is_logged_and_leaves_the_file_unlinked(self):
+        self.metadata = {"prompt": sampled_graph("a quiet harbor", "blurry")}
+        path = self.image("plain_00003_.png")
+        with (
+            mock.patch(
+                "utils.foreign_runs.ingest_resolved_pair",
+                side_effect=sqlite3.OperationalError("locked"),
+            ),
+            self.assertLogs("prompt_manager.image_monitor", "ERROR"),
+        ):
+            self.handler.process_new_image(path)
+        self.assertIsNone(self.linked_text(path))
+
+
 class TestBoundedLookup(RecordedRuns, unittest.TestCase):
     """M5: history lookups are batched, and the history schema is unchanged."""
 

tests/test_settings_sources.py

@@ -323,6 +323,43 @@ class TestMonitoringSetting(SourcesTestCase):
         self.assertIs(GalleryConfig.MONITORING_ENABLED, False)
 
 
+class TestSaveForeignRunsSetting(SourcesTestCase):
+    """#247 D-Q4: "Save prompts from other workflows", applied live."""
+
+    async def asyncSetUp(self):
+        await super().asyncSetUp()
+        saved = GalleryConfig.SAVE_FOREIGN_RUNS
+        self.addCleanup(setattr, GalleryConfig, "SAVE_FOREIGN_RUNS", saved)
+        GalleryConfig.SAVE_FOREIGN_RUNS = True
+
+    async def test_read_and_write_published_live_without_a_restart(self):
+        self.assertIs((await self._get())["settings"]["save_foreign_runs"], True)
+        self._install_monitor(FakeMonitor([self.output_dir]))
+        status, data = await self._post(
+            "/prompt_manager/settings", {"save_foreign_runs": False}
+        )
+        self.assertEqual(status, 200, data)
+        self.assertTrue(data["persisted"])
+        self.assertEqual(data["restart_required_fields"], [])
+        saved = self._read_config()
+        self.assertIs(saved["gallery"]["monitoring"]["save_foreign_runs"], False)
+        self.assertIs(GalleryConfig.SAVE_FOREIGN_RUNS, False)
+        self.assertIs((await self._get())["settings"]["save_foreign_runs"], False)
+
+    async def test_only_true_or_false_is_accepted(self):
+        self._write_config({"custom": {"keep": True}})
+        before = self._config_bytes()
+        for value in ("false", 0, None, []):
+            with self.subTest(value=value):
+                status, data = await self._post(
+                    "/prompt_manager/settings", {"save_foreign_runs": value}
+                )
+                self.assertEqual(status, 400, data)
+                self.assertIn("save_foreign_runs must be true or false", data["error"])
+                self.assertEqual(self._config_bytes(), before)
+                self.assertIs(GalleryConfig.SAVE_FOREIGN_RUNS, True)
+
+
 class TestSourceChangeHold(SourcesTestCase):
     async def test_root_save_refused_while_scan_job_runs(self):
         GalleryConfig.MONITORING_DIRECTORIES = [self._root(self.a)]

tests/vendor/comfyui-config.test.mjs

@@ -36,14 +36,14 @@ test('execution script discovers only the new matrix and preserves all 78 origin
     // + the fresh-host library location case, + #342 image attribution, + #343 linked text,
     // + #332 image category routes on a generated output (image-attribution.spec.mjs, 2 projects),
     // + #347 cached runtime-linked runs: five node kinds and eight one-renderer cases (execution-text-link.spec.mjs, 2 projects),
-    // + #90 Save to library off (five node types)
-    assert.match(executions, /Total: 128 tests in 5 files/);
+    // + #90 Save to library off (five node types), + #247 foreign workflows (image-attribution.spec.mjs, 2 tests x 2 projects)
+    assert.match(executions, /Total: 132 tests in 5 files/);
     for (const project of ['classic-start', 'nodes2-start']) {
         const cases = executions.split('\n').filter(line => line.includes(`[${project}]`));
         assert.equal(cases.filter(line => line.includes('execution-history.spec.mjs')).length, 15);
         assert.equal(cases.filter(line => line.includes('execution-outputs.spec.mjs')).length, 8);
         assert.equal(cases.filter(line => line.includes('library-location.spec.mjs')).length, 1);
-        assert.equal(cases.filter(line => line.includes('image-attribution.spec.mjs')).length, 3);
+        assert.equal(cases.filter(line => line.includes('image-attribution.spec.mjs')).length, 5);
         assert.equal(cases.filter(line => line.includes('execution-text-link.spec.mjs')).length, 37);
     }
     assert.doesNotMatch(executions, /prompt-manager.spec/);

utils/foreign_runs.py

@@ -0,0 +1,142 @@
+"""The monitor's path for outputs of workflows without a PromptManager node (#247).
+
+A foreign output (no positive PromptManager node in its graph) is decided by
+the ladder the output scan uses (``utils.file_prompt.decide_file_prompt``)
+with the ``MONITOR`` policy: certain files are stored at once, files only
+recorded history can decide wait for their run's outputs to be sealed, and
+everything else stays unlinked. This module holds the parts that do not
+need the monitor's waiting sets: the setting, the decision call and the
+store. The waiting and the re-check stay on ``ImageGenerationHandler``.
+"""
+
+import os
+from dataclasses import dataclass
+from typing import Any, Callable, Mapping, Optional, Tuple
+
+from .file_prompt import (
+    NOT_SAVED,
+    NOT_SAVED_WIRED,
+    PAIR,
+    POSITIVE,
+    Decision,
+    Policy,
+    decide_file_prompt,
+)
+from .hashing import generate_prompt_hash
+from .usage_tracking import survive_switch
+
+try:
+    from ..database.pair_ingestion import RUN_MARKS, ingest_resolved_pair
+    from ..database.run_attribution import recorded_output_lookup
+    from ..database.scanned_prompts import store_scanned_prompt_or_raise
+except ImportError:
+    from database.pair_ingestion import RUN_MARKS, ingest_resolved_pair
+    from database.run_attribution import recorded_output_lookup
+    from database.scanned_prompts import store_scanned_prompt_or_raise
+
+# Decisions that name a prompt to store
+CERTAIN = (PAIR, POSITIVE)
+# Decisions of a run whose prompt has Save to library off, or wired from a
+# node the file cannot show (#90): final, nothing is stored, nothing waits
+NOT_SAVED_KINDS = (NOT_SAVED, NOT_SAVED_WIRED)
+# Sources certain without a recorded run: the graph alone, or its recorded saver
+ANCHORED_SOURCES = ("inspector", "recorded-run")
+
+
+@dataclass(frozen=True)
+class ForeignRecheck:
+    """``schedule`` snapshot: the one re-check of a foreign output after its
+    run's outputs were sealed. ``reference`` is the sealed output reference,
+    the spelling history recorded (``recorded_output_lookup(reference=)``)."""
+
+    reference: Mapping[str, Any]
+
+
+def save_foreign_runs() -> bool:
+    """Whether a foreign output may be stored now, read for every file.
+
+    ``GalleryConfig.SAVE_FOREIGN_RUNS`` and ``MONITORING_ENABLED``: a
+    monitoring switch saved off stops foreign writes at once, although the
+    running observer watches until a restart. True when the configuration
+    cannot be imported (the defaults).
+    """
+    try:
+        from ..py.config import GalleryConfig
+    except ImportError:
+        try:
+            from py.config import GalleryConfig
+        except ImportError:
+            return True
+    return (
+        GalleryConfig.SAVE_FOREIGN_RUNS is not False
+        and GalleryConfig.MONITORING_ENABLED is not False
+    )
+
+
+def history_lookup(
+    db: Any, image_path: str, reference: Optional[Mapping[str, Any]] = None
+) -> Callable[[], Any]:
+    """The ladder's ``lookup`` for one file: its ``RecordedOutput`` or None.
+
+    With ``reference`` the file is spelled as history recorded it. Failures
+    are logged by ``recorded_output_lookup`` and name nothing.
+    """
+    lookup = recorded_output_lookup(
+        db, reference=dict(reference) if reference is not None else None
+    )
+    return lambda: lookup(image_path)
+
+
+def decide_foreign(
+    image_path: str,
+    metadata: Any,
+    *,
+    in_output: bool,
+    lookup: Optional[Callable[[], Any]],
+) -> Decision:
+    """The ``MONITOR`` ladder for one foreign output (the size is read only
+    when the ladder needs it)."""
+
+    def file_size() -> Optional[int]:
+        try:
+            return os.path.getsize(image_path)
+        except OSError:
+            return None
+
+    return decide_file_prompt(
+        metadata,
+        file_name=os.path.basename(image_path),
+        file_size=file_size,
+        in_output=in_output,
+        lookup=lookup,
+        policy=Policy.MONITOR,
+    )
+
+
+def store_foreign(
+    db: Any, image_path: str, decision: Decision
+) -> Tuple[str, Optional[int]]:
+    """Store a certain decision: ``(status, prompt id)``.
+
+    One write through ``survive_switch``, so it runs again on the moved
+    library when a Move library switched while it waited (#249 S6b). Created
+    prompts carry ``RUN_MARKS``. A file linked to another prompt is a
+    ``conflict`` and nothing is written. Database errors are raised.
+    """
+
+    def store():
+        if decision.kind == PAIR:
+            result = ingest_resolved_pair(
+                db, image_path, decision.canonical, decision.resolution, marks=RUN_MARKS
+            )
+            return result["status"], result.get("image_id")
+        return store_scanned_prompt_or_raise(
+            db, image_path, decision.canonical, decision.text, RUN_MARKS
+        )
+
+    status, linked = survive_switch(db, store)
+    if decision.kind != PAIR or linked is None:
+        return status, linked
+    # The pair ingester reports the image row; its prompt is the positive's.
+    prompt = db.get_prompt_by_hash(generate_prompt_hash(decision.text))
+    return status, prompt["id"] if prompt else None

utils/image_monitor.py

@@ -33,6 +33,17 @@ from typing import Callable, Optional, Dict, Any, List, Tuple
 from watchdog.observers import Observer
 from watchdog.events import FileSystemEventHandler
 
+from .file_prompt import HISTORY, NOT_SAVED, NOT_SAVED_WIRED
+from .foreign_runs import (
+    ANCHORED_SOURCES,
+    CERTAIN,
+    NOT_SAVED_KINDS,
+    ForeignRecheck,
+    decide_foreign,
+    history_lookup,
+    save_foreign_runs,
+    store_foreign,
+)
 from .hashing import generate_prompt_hash
 from .metadata_extractor import ComfyUIMetadataExtractor
 from .parallel import note_refusal, refused_at_shutdown
@@ -153,6 +164,11 @@ class ImageGenerationHandler(FileSystemEventHandler):
         # Sealed re-checks ``schedule`` refused (the path still pending, or the
         # pending set full): retried each time a pending image finishes.
         self._history_due: "OrderedDict[str, None]" = OrderedDict()
+        # The same two sets for foreign outputs (#247), apart so a large
+        # foreign run never evicts a waiting keyless file; the due set keeps
+        # each file's sealed output reference.
+        self._awaiting_foreign: "OrderedDict[str, Tuple[str, float]]" = OrderedDict()
+        self._foreign_due: "OrderedDict[str, dict]" = OrderedDict()
         self._awaiting_lock = threading.Lock()
         self._sealed_subscribed = False
         # Set by ``close``: no new work, no new worker, ever (#302).
@@ -253,6 +269,8 @@ class ImageGenerationHandler(FileSystemEventHandler):
         with self._awaiting_lock:
             self._awaiting.clear()
             self._history_due.clear()
+            self._awaiting_foreign.clear()
+            self._foreign_due.clear()
         with self._pending_lock:
             self._pending.clear()
             worker, self._worker = self._worker, None
@@ -438,6 +456,9 @@ class ImageGenerationHandler(FileSystemEventHandler):
             if prompt_snapshot is HISTORY_ONLY:
                 self._recheck_history(image_path)
                 return
+            if isinstance(prompt_snapshot, ForeignRecheck):
+                self._recheck_foreign(image_path, prompt_snapshot.reference)
+                return
             self.logger.info(f"Processing image: {image_path}")
 
             if not os.path.exists(image_path):
@@ -457,17 +478,12 @@ class ImageGenerationHandler(FileSystemEventHandler):
             # negative-prompt PromptManager nodes in the workflow.
             state, current_prompt = self._find_prompt_from_metadata(metadata)
             if state == "opted_out":
-                self.logger.info(
-                    "Save to library is off for this run's prompt, skipping image: "
-                    f"{image_path}"
-                )
+                self._log_not_saved(image_path)
                 return
             if state == "foreign":
-                # Another workflow saved this image: it belongs to no prompt of
-                # ours, and must not consume a batch queue entry.
-                self.logger.info(
-                    f"No PromptManager prompt in workflow, skipping image: {image_path}"
-                )
+                # Another workflow saved this image: never a batch queue entry
+                # of ours; its own graph may name its prompt (#247).
+                self._ingest_foreign(image_path, metadata)
                 return
             if current_prompt:
                 self.logger.info(
@@ -589,36 +605,55 @@ class ImageGenerationHandler(FileSystemEventHandler):
             resolved = os.path.abspath(image_path)
         return os.path.normcase(resolved)
 
-    def _prune_awaiting(self) -> None:
-        """Drop expired and surplus entries (``_awaiting_lock`` held)."""
+    def _prune_awaiting(self, store=None) -> List[Tuple[str, bool]]:
+        """Drop expired and surplus entries of ``store`` (the keyless set by
+        default; ``_awaiting_lock`` held). Returns the dropped foreign files
+        as ``(path, evicted)`` for ``_log_dropped`` after the lock."""
+        store = self._awaiting if store is None else store
         now = time.monotonic()
-        for key in [k for k, (_p, due) in self._awaiting.items() if due <= now]:
-            del self._awaiting[key]
-        while len(self._awaiting) > MAX_AWAITING_HISTORY:
-            self._awaiting.popitem(last=False)
+        expired = [k for k, (_p, due) in store.items() if due <= now]
+        dropped = [(store.pop(key)[0], False) for key in expired]
+        while len(store) > MAX_AWAITING_HISTORY:
+            dropped.append((store.popitem(last=False)[1][0], True))
+        return dropped if store is self._awaiting_foreign else []
 
-    def _await_history(self, image_path: str) -> None:
+    def _log_dropped(self, dropped: List[Tuple[str, bool]]) -> None:
+        """Log foreign files that stopped waiting (outside ``_awaiting_lock``)."""
+        for image_path, evicted in dropped:
+            name = os.path.basename(image_path)
+            if evicted:
+                self.logger.info(
+                    f"Too many files wait for their run; {name} left unlinked"
+                )
+            else:
+                self.logger.info(f"No recorded run for {name}; left unlinked")
+
+    def _await_history(self, image_path: str, store=None) -> None:
         """Re-check ``image_path`` once, history only, when its run is sealed.
 
-        A closed handler re-checks nothing.
+        ``store`` is the waiting set (``_awaiting_foreign`` for a foreign
+        output). A closed handler re-checks nothing.
         """
         if self._closed:
             return
         if not self._sealed_subscribed:
             subscribe_outputs_sealed(self._on_outputs_sealed)
             self._sealed_subscribed = True
+        store = self._awaiting if store is None else store
         key = self._awaiting_key(image_path)
         with self._awaiting_lock:
-            self._awaiting.pop(key, None)
-            self._awaiting[key] = (
-                image_path,
-                time.monotonic() + AWAIT_HISTORY_SECONDS,
-            )
-            self._prune_awaiting()
+            store.pop(key, None)
+            store[key] = (image_path, time.monotonic() + AWAIT_HISTORY_SECONDS)
+            dropped = self._prune_awaiting(store)
+        self._log_dropped(dropped)
 
-    def _stop_awaiting(self, image_path: str) -> None:
+    def _await_foreign(self, image_path: str) -> None:
+        self._await_history(image_path, self._awaiting_foreign)
+
+    def _stop_awaiting(self, image_path: str, store=None) -> None:
+        store = self._awaiting if store is None else store
         with self._awaiting_lock:
-            self._awaiting.pop(self._awaiting_key(image_path), None)
+            store.pop(self._awaiting_key(image_path), None)
 
     def awaiting_history(self) -> List[str]:
         """Normalized paths still waiting for their run's recorded outputs."""
@@ -626,6 +661,14 @@ class ImageGenerationHandler(FileSystemEventHandler):
             self._prune_awaiting()
             return list(self._awaiting)
 
+    def awaiting_foreign(self) -> List[str]:
+        """Normalized paths of foreign outputs waiting for their run (#247)."""
+        with self._awaiting_lock:
+            dropped = self._prune_awaiting(self._awaiting_foreign)
+            waiting = list(self._awaiting_foreign)
+        self._log_dropped(dropped)
+        return waiting
+
     def _on_outputs_sealed(self, run_id: str, references: List[dict]) -> None:
         """A run's outputs were recorded: re-check its waiting images once.
 
@@ -637,6 +680,7 @@ class ImageGenerationHandler(FileSystemEventHandler):
         due = []
         with self._awaiting_lock:
             self._prune_awaiting()
+            dropped = self._prune_awaiting(self._awaiting_foreign)
             for reference in references:
                 if reference.get("folder_type") != "output":
                     continue
@@ -647,39 +691,53 @@ class ImageGenerationHandler(FileSystemEventHandler):
                 )
                 waiting = self._awaiting.pop(key, None)
                 if waiting is not None:
-                    due.append(waiting[0])
-        for image_path in due:
-            self._schedule_recheck(image_path)
+                    due.append((waiting[0], None))
+                waiting = self._awaiting_foreign.pop(key, None)
+                if waiting is not None:
+                    due.append((waiting[0], dict(reference)))
+        self._log_dropped(dropped)
+        for image_path, reference in due:
+            self._schedule_recheck(image_path, reference)
 
-    def _schedule_recheck(self, image_path: str) -> None:
-        """Schedule the history-only re-check, or keep it due for later.
+    @staticmethod
+    def _recheck_snapshot(reference):
+        return HISTORY_ONLY if reference is None else ForeignRecheck(reference)
+
+    def _schedule_recheck(self, image_path: str, reference=None) -> None:
+        """Schedule the history-only re-check (with the sealed ``reference``:
+        a foreign output's), or keep it due for later.
 
         A closed handler drops it.
         """
         if self._closed:
             return
-        if self.schedule(image_path, HISTORY_ONLY):
+        if self.schedule(image_path, self._recheck_snapshot(reference)):
             return
         with self._awaiting_lock:
             # ``close`` sets the flag before ``shutdown`` clears this set, so
             # a close that raced the refusal above is seen here.
             if self._closed:
                 return
-            self._history_due.pop(image_path, None)
-            self._history_due[image_path] = None
-            while len(self._history_due) > MAX_AWAITING_HISTORY:
-                self._history_due.popitem(last=False)
+            due = self._history_due if reference is None else self._foreign_due
+            due.pop(image_path, None)
+            due[image_path] = reference
+            evicted = []
+            while len(due) > MAX_AWAITING_HISTORY:
+                evicted.append(due.popitem(last=False)[0])
+        if reference is not None:
+            self._log_dropped([(path, True) for path in evicted])
 
     def _drain_history_due(self) -> None:
         """Schedule due re-checks now that a pending slot or path is free."""
         with self._awaiting_lock:
-            if not self._history_due:
-                return
-            due = list(self._history_due)
-        for image_path in due:
-            if self.schedule(image_path, HISTORY_ONLY):
+            due = [(self._history_due, p, None) for p in self._history_due]
+            due += [(self._foreign_due, *item) for item in self._foreign_due.items()]
+        for store, image_path, reference in due:
+            if self.schedule(image_path, self._recheck_snapshot(reference)):
                 with self._awaiting_lock:
-                    self._history_due.pop(image_path, None)
+                    # A newer seal may have replaced it meanwhile: keep that one
+                    if image_path in store and store[image_path] is reference:
+                        del store[image_path]
 
     def _recheck_history(self, image_path: str) -> None:
         """The one history-only re-check of a keyless image (#342)."""
@@ -703,6 +761,108 @@ class ImageGenerationHandler(FileSystemEventHandler):
             metadata or {"file_info": self.get_basic_file_info(image_path)},
         )
 
+    def _ingest_foreign(self, image_path: str, metadata) -> None:
+        """A foreign output at settle: store its certain prompt, or wait for
+        its run's recorded outputs when only history can decide (#247)."""
+        name = os.path.basename(image_path)
+        if not save_foreign_runs():
+            self.logger.info(
+                f"No PromptManager prompt in workflow, skipping image: {image_path}"
+            )
+            return
+        in_output = in_output_folder(image_path, host_output_directory())
+        decision = decide_foreign(
+            image_path, metadata, in_output=in_output, lookup=None
+        )
+        if decision.kind == HISTORY:
+            decision = self._settle_lookup(image_path, metadata)
+            if decision is None:
+                self.logger.debug(f"{name} waits for its run to end")
+                return
+            self._stop_awaiting(image_path, self._awaiting_foreign)
+        if decision.kind in NOT_SAVED_KINDS:
+            self._log_not_saved(image_path, decision.kind)
+        elif decision.kind in CERTAIN:
+            self._store_foreign(image_path, decision)
+        else:
+            self.logger.info(f"No certain prompt for {name}; left unlinked")
+
+    def _log_not_saved(self, image_path: str, kind: str = NOT_SAVED) -> None:
+        """The log line of a file whose run's prompt does not save (#90)."""
+        if kind == NOT_SAVED_WIRED:
+            reason = "is wired from a node the image cannot show"
+        else:
+            reason = "is off"
+        self.logger.info(
+            f"Save to library {reason} for this run's prompt, skipping image: "
+            f"{image_path}"
+        )
+
+    def _settle_lookup(self, image_path: str, metadata):
+        """The settle-time history lookup of a file only history can decide.
+
+        Registered first, so a run sealed before this lookup is not missed.
+        Returns None to keep waiting: no record yet, and the answer is not
+        the inspector's or the recorded saver's (the legacy floor waits for
+        the re-check). With a record, any decision ends the wait. A
+        ``not_saved`` here is the undecided step's (an early one never asks
+        history, so it never reaches this lookup): a recorded saver may still
+        resolve another branch, so without a record it waits too (#90).
+        """
+        self._await_foreign(image_path)
+        found = []
+        lookup = history_lookup(self.db_manager, image_path)
+
+        def recorded():
+            found.append(lookup())
+            return found[-1]
+
+        decision = decide_foreign(image_path, metadata, in_output=True, lookup=recorded)
+        if any(record is not None for record in found):
+            return decision
+        if decision.kind in CERTAIN and decision.source in ANCHORED_SOURCES:
+            return decision
+        return None
+
+    def _recheck_foreign(self, image_path: str, reference) -> None:
+        """The one re-check of a foreign output after its run ended (#247)."""
+        name = os.path.basename(image_path)
+        if not os.path.exists(image_path):
+            return
+        if not save_foreign_runs():
+            self.logger.info(f"Saving prompts from other workflows is off: {name}")
+            return
+        try:
+            metadata = self.metadata_extractor.extract_metadata(image_path)
+        except Exception as error:
+            self.logger.warning(f"Metadata of {name} could not be read: {error}")
+            return
+        lookup = history_lookup(self.db_manager, image_path, reference)
+        decision = decide_foreign(image_path, metadata, in_output=True, lookup=lookup)
+        if decision.kind in NOT_SAVED_KINDS:
+            self._log_not_saved(image_path, decision.kind)
+            return
+        if decision.kind in CERTAIN:
+            self._store_foreign(image_path, decision)
+            return
+        self.logger.info(
+            f"No certain prompt for {name} after its run ended; left unlinked"
+        )
+
+    def _store_foreign(self, image_path: str, decision) -> None:
+        name = os.path.basename(image_path)
+        if not save_foreign_runs():  # switched off while this file was decided
+            self.logger.info(f"Saving prompts from other workflows is off: {name}")
+            return
+        try:
+            status, prompt_id = store_foreign(self.db_manager, image_path, decision)
+        except Exception as error:
+            self.logger.error(f"Failed to store the prompt of {name}: {error}")
+            return
+        self.logger.info(
+            f"Foreign run: {status} prompt {prompt_id} for {name} ({decision.source})"
+        )
+
     def _history_prompt(self, image_path: str) -> Optional[int]:
         """The stored prompt recorded history attributes to this file, if one."""
         try:

web/js/workspace/system-sources.mjs

@@ -14,6 +14,9 @@
  *   it to the host draft of `gallery_root_paths`; Remove takes one out. Both wait for the one Save row.
  * - "Watch for new outputs" (`monitoring_enabled`) and "Worker threads" (`worker_threads`, a range up to the server's
  *   `cpu_count`, D113) are host drafts too. "Enrich existing metadata" reviews a scan of every source.
+ * - D158 (#247): "Save prompts from other workflows" (`save_foreign_runs`), a host draft below "Watch for new outputs",
+ *   applied by the server at once (the monitor reads it for every file); disabled, saying why, while monitoring is off,
+ *   unless the running monitor still watches until a restart (D128 `stopping`): then it can still be set.
  * - "Thumbnail generation → Review options" opens "Thumbnail maintenance" (thumbnail-review.mjs, S5 Task 14), which
  *   starts the thumbnail Activity job.
  *
@@ -63,6 +66,21 @@ export const LOAD_FAILED_TEXT = 'Sources could not be loaded. Your saved data ha
 export const DUPLICATE_TEXT = 'This folder is already a source.';
 export const VALIDATE_FAILED_TEXT = 'The folder could not be checked. Try again.';
 export const REFRESH_FAILED_TEXT = 'Files could not be refreshed. Try again.';
+/** D158 (#247): the row the concept does not draw, below "Watch for new outputs". */
+export const FOREIGN_RUNS = Object.freeze({ id: 'foreign-runs', key: 'save_foreign_runs', label: 'Save prompts from other workflows',
+    description: 'New outputs of workflows without a PromptManager node add their prompt to the library when it is certain.',
+    off: 'Applies while Watch for new outputs is on.' });
+
+/**
+ * D158: the row for the drafted `save_foreign_runs` (`value`) and `monitoring_enabled` (`monitoring`): disabled until
+ * read, and while monitoring is off, when `note` says why (the value is kept and saved as it is). `stopping`: a source
+ * is still watched until the restart (D128), so the value can still be set; the note stays.
+ */
+export function foreignRunsState({ value, monitoring, stopping = false }) {
+    if (value === undefined) return { disabled: true, checked: null, note: null };
+    const off = monitoring === false;
+    return { disabled: off && !stopping, checked: value === true, note: off ? FOREIGN_RUNS.off : null };
+}
 
 // Labels of the sources added in this page session (validate-source answers them), for the rows not saved yet.
 const addedLabels = new Map();
@@ -267,13 +285,16 @@ export function mountSources(container, { api, activity, scope, drafts, notify =
     const monitoring = settingRow({ id: 'monitoring', label: 'Watch for new outputs',
         description: 'New files enter the same library. Existing scans do not count as new runs.', control: { type: 'toggle' },
         scope, onChange: value => drafts.set('host', 'monitoring_enabled', value) });
+    const foreign = settingRow({ id: FOREIGN_RUNS.id, label: FOREIGN_RUNS.label, description: FOREIGN_RUNS.description,
+        control: { type: 'toggle' }, scope, onChange: value => drafts.set('host', FOREIGN_RUNS.key, value) });
     const workers = settingRow({ id: 'workers', label: 'Worker threads',
         description: 'Balance background scans and thumbnail work against generation.', control: { type: 'range', value: 4 },
         scope, onChange: value => drafts.set('host', 'worker_threads', value) });
-    const hostRows = [[monitoring, 'monitoring', 'monitoring_enabled'], [workers, 'workers', 'worker_threads']]
+    const hostRows = [[monitoring, 'monitoring', 'monitoring_enabled'], [foreign, FOREIGN_RUNS.id, FOREIGN_RUNS.key],
+        [workers, 'workers', 'worker_threads']]
         .map(([row, id, key]) => ({ input: controlOf(row, id), key, text: row.children[0], note: null }));
     for (const row of hostRows) row.input.dataset.key = row.key;
-    surface.append(monitoring, workers,
+    surface.append(monitoring, foreign, workers,
         operationRow({ scope, title: 'Enrich existing metadata', description: 'Resolve both prompts, deduplicate recipes, preserve uncertain links.',
             label: 'Review scan', onClick: scanAll }),
         operationRow({ scope, title: 'Thumbnail generation', description: 'Generate missing previews; rebuilding all is a separate explicit option.',
@@ -348,6 +369,10 @@ export function mountSources(container, { api, activity, scope, drafts, notify =
         for (const row of hostRows) {
             const value = drafts.get('host', row.key);
             row.input.disabled = value === undefined;
+            if (row.key === FOREIGN_RUNS.key) {
+                paintForeign(row, value, unavailable);
+                continue;
+            }
             if (row.key === 'worker_threads') {
                 // D113: the range ends at the server's core count (the concept's fixed 16).
                 const cpus = drafts.get('host', 'cpu_count');
@@ -360,6 +385,24 @@ export function mountSources(container, { api, activity, scope, drafts, notify =
         }
     }
 
+    // D158: disabled while monitoring is drafted off, with the reason under the description.
+    function paintForeign(row, value, unavailable) {
+        const stopping = sources?.effective?.some(source => source.watch === 'stopping') === true;
+        const state = foreignRunsState({ value, monitoring: drafts.get('host', 'monitoring_enabled'), stopping });
+        row.input.disabled = state.disabled;
+        if (state.checked !== null) row.input.checked = state.checked;
+        const note = unavailable && value === undefined ? HOST_UNAVAILABLE : state.note;
+        if (note && row.note?.textContent !== note) {
+            row.note?.remove();
+            row.note = el('p', note, 'small');
+            row.note.id = `${FOREIGN_RUNS.id}-note`;
+            row.text.append(row.note);
+        } else if (!note && row.note) { row.note.remove(); row.note = null; }
+        // The note is the switch's description, so a screen reader says why it is disabled.
+        if (row.note) row.input.setAttribute('aria-describedby', row.note.id);
+        else row.input.removeAttribute('aria-describedby');
+    }
+
     // The sources changed since this page read them (a refused per-source scan): read them and the saved values again.
     function reread() {
         load();
@@ -378,6 +421,7 @@ export function mountSources(container, { api, activity, scope, drafts, notify =
         if (answer === STALE || scope.signal.aborted) return;
         sources = answer?.sources ?? null;
         paintList();
+        paintControls();
     }
 
     // A save (or a reread after a conflict) changed the saved sources or monitoring: read their state again. The first