diff --git a/packages/core/src/opendirector/application/creation_history.py b/packages/core/src/opendirector/application/creation_history.py index b99c5e73..fe5ebba5 100644 --- a/packages/core/src/opendirector/application/creation_history.py +++ b/packages/core/src/opendirector/application/creation_history.py @@ -467,6 +467,9 @@ def _inventory(self, snapshot, *, records=None): for doc in snapshot.editorDocuments: if doc.kind == 'subject': current.update(ref.assetId for ref in asset_references(doc)) + # A live pinned reference keeps its fixed media reusable even without a canvas + # node, after the media has also become a generated result's source. + current.update(ref.source.assetId for ref in snapshot.references if ref.active and ref.binding == 'pinned') return current, current_results, historical def asset_list(self, project_id, *, text='', usage='all', visibility='all', category='all', offset=0, limit=100): diff --git a/tests/test_creation_history_service.py b/tests/test_creation_history_service.py index 58f01b72..ab205615 100644 --- a/tests/test_creation_history_service.py +++ b/tests/test_creation_history_service.py @@ -414,3 +414,43 @@ def test_asset_categories_include_subject_and_timeline_nodes_with_counts_before_ assert empty['total'] == 1 and empty['nodes'][0]['asset'] is None assert empty['nodes'][0]['itemId'] == empty_item and not empty['nodes'][0]['visible'] assert service.asset_list(pid, category='subject', text='empty', visibility='visible')['total'] == 0 + + +def test_asset_used_only_by_an_active_pinned_reference_stays_listed_after_becoming_a_source(project): + from opendirector.application.creation_canvas import asset_ref + store, pid, session = project + assert send(store, pid, 'asset.writeText', session=session, name='pinned.md', text='fixed').status == 'applied' + source = store.snapshot(pid).assets[-1] + target_item, spec = media(project, title='Target') + reference = dict(id=uuid4(), projectId=pid, targetSpecId=spec.id, binding='pinned', + source=asset_ref(source).model_dump(mode='json'), order=0) + assert send(store, pid, 'reference.add', session=session, + versions=[ver('generation_spec', spec), ver('canvas_item', target_item)], + reference=reference).status == 'applied' + service = CreationHistory(store) + def listed(**kwargs): + return {row['asset'].id for row in service.asset_list(pid, **kwargs)['items']} + def catalog(): + return next(row for row in service.reference_catalog(pid) if row['id'] == str(source.id)) + assert source.id in listed() and source.id in listed(visibility='hidden') + assert catalog()['available'] + # Any newer asset naming it in provenance.sources, like a generated result, makes it historical. + assert send(store, pid, 'asset.writeText', session=session, versions=[ver('asset', source)], name='derived.md', + text='derived', previousAsset=asset_ref(source).model_dump(mode='json')).status == 'applied' + assert store.snapshot(pid).assets[-1].provenance.sources == (asset_ref(source),) + # Becoming another asset's source must not hide media a live configuration still pins. + assert source.id in listed() and source.id in listed(visibility='hidden') + assert catalog()['available'] and catalog()['fileAvailable'] + reopened = CreationHistory(CreationProjects(LocalProjects(store.local.repository))) + assert source.id in {row['asset'].id for row in reopened.asset_list(pid, visibility='hidden')['items']} + assert service.usage(store.snapshot(pid), source.id)['used'] + current = next(item for item in store.snapshot(pid).references if item.id == reference['id']) + target = next(item for item in store.snapshot(pid).generationSpecs if item.id == spec.id) + assert send(store, pid, 'reference.remove', session=session, + versions=[ver('reference', current), ver('generation_spec', target)], + referenceId=current.id).status == 'applied' + # Once removed, it is again only a historical source, as before. + assert source.id not in listed() + assert not catalog()['available'] + # Delete protection is unchanged: the derived version and history still depend on it. + assert service.usage(store.snapshot(pid), source.id)['used']